столбец данных pyspark: столбец Hive

Question

У меня есть одна таблица Hive:

hive> describe stock_quote;
OK
tickerid                string                                      
tradeday                string                                      
tradetime               string                                      
openprice               string                                      
highprice               string                                      
lowprice                string                                      
closeprice              string                                      
volume                  string

Следующий код Spark читает файлы CSV и пытается вставить записи в таблицу Hive:

sc = spark.sparkContext
lines = sc.textFile('file:///<File Location>')
rows = lines.map(lambda line : line.split(','))
rows_map = rows.map(lambda row : Row(TickerId = row[0], TradeDay = row[1], TradeTime = row[2], OpenPrice = row[3], HighPrice = row[4], LowPrice = row[5], ClosePrice = row[6], Volume = row[7]))
rows_df = spark.createDataFrame(rows_map)
rows_df.write.mode('append').insertInto('default.stock_quote')

Проблема, с которой я сталкиваюсь,что, когда я вызываю функцию show () для фрейма данных, он печатает столбцы в алфавитном порядке, как показано ниже

|ClosePrice|HighPrice|LowPrice|OpenPrice|TickerId|TradeDay|TradeTime|Volume|

, а в таблицу вставляет значение ClosePrice (1-й столбец в DF) в TickerId (1-й столбец)в столбце Hive), значение HighPrice в столбце TradeDay и т. д.

Попытка вызова функции select () на фрейме данных не помогла.Попытался поместить список имен столбцов следующим образом:

rows_df = spark.createDataFrame(rows_map, ["TickerId", "TradeDay", "TradeTime", "OpenPrice", "HighPrice", "LowPrice", "ClosePrice", "Volume"])

Приведенный выше изменяет порядок имен столбцов, но значения остаются на той же позиции, что еще более неправильно.

Любая помощь будет действительнооценили.

Answer 1

Вы должны пойти с namedtuple вместо Row, потому что 'Row' пытается упорядочить имена столбцов.Таким образом, упорядоченные имена столбцов не совпадают с порядком столбцов таблицы default.stock_quote. Проверьте Что такое эквивалент класса случая Scala в PySpark? для получения более подробной информации

Итак, выдолжен делать

from collections import namedtuple

table = namedtuple('table', ['TickerId', 'TradeDay', 'TradeTime', 'OpenPrice', 'HighPrice', 'LowPrice', 'ClosePrice', 'Volume'])
rows_map = rows.map(lambda row : table(row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7]))

И, как подсказал @ user6910411, " нормальный кортеж тоже подойдет "

rows_map = rows.map(lambda row : (row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7]))
rows_df = spark.createDataFrame(rows_map, ['TickerId', 'TradeDay', 'TradeTime', 'OpenPrice', 'HighPrice', 'LowPrice', 'ClosePrice', 'Volume'])

теперь insertIntoдолжно работать

Answer 2

Вы также можете использовать saveAsTable вместо insertInto

Из документов :

В отличие от insertInto, saveAsTable будет использоватьимена столбцов, чтобы найти правильные положения столбцов

Answer 3

Как получилось, отсортировано в алфавитном порядке?Так ли это в файле csv?

В любом случае, я бы сделал это следующим образом:

• выберите столбцы из вашей таблицы
• измените порядок данных на основе столбцовиз таблицы

# pyspark below
list_columns = spark.sql('select * from table').columns # there might be simpler way
dataframe.select(*list_columns)