Загрузка выбранного столбца из CSV-файла в фрейм данных в Spark - PullRequest
Новая
       10

Загрузка выбранного столбца из CSV-файла в фрейм данных в Spark

1 голос
/ 27 апреля 2020

Я пытаюсь загрузить CSV-файл в информационный фрейм Spark. Файл CSV не имеет заголовка как такового, но я знаю, какое поле чему соответствует. Проблема в том, что мой CSV имеет почти 35 нечетных полей, но меня интересуют очень ограниченные столбцы, поэтому есть способ, с помощью которого я могу загрузить выбранные столбцы и отобразить их в соответствующие поля, как определено в моей схеме.

Давайте скажем, у нас есть следующий CSV: 

1,Michel,1256,Student,high Street, New Delhi
2,Solace,7689,Artist,M G Road, Karnataka

В Scala мой код выглядит примерно так. 

val sample_schema = StructType(Array(StructField("Name", StringType, nullable = false),
      StructField("unique_number", StringType, nullable = false),
      StructField("state", StringType, nullable = false))
val blogsDF = sparkSession.read.schema(sample_schema)
                              .option("header", true)
                              .csv(file_path)

Это загрузит данные в фрейм данных, но будет не в том порядке, в котором я хочу. Я хочу, чтобы csv-запись была разделена и данные загружались в соответствии с базовым отображением 

col1 --> Name
col2 --> unique id
col5 --> state

Не уверен, что мы можем выполнить такую ​​операцию перед загрузкой данных в DataFrame. Я знаю другой подход, при котором мы можем загрузить данные в один фрейм данных, а затем выбрать несколько столбцов и создать другой фрейм данных, просто хочу проверить, можем ли мы отобразить карту во время самой загрузки данных. Любая помощь или указатель в этом отношении будут действительно полезны.

Спасибо Ashit

Ответы [ 2 ]

0 голосов
/ 27 апреля 2020

В Spark 2.4 опция конфигурации spark.sql.csv.parser.columnPruning.enabled предназначена для чего-то очень похожего - она ​​позволяет загружать только столбцы, указанные выбором (а не схемой). По умолчанию установлено значение true, см. Руководство по обновлению Spark . Вы должны определить «полную» схему (т.е. все ваши 35+ столбцов), чтобы помочь парсеру CSVDataSource перемещаться по вводу. Так что-то вроде этого должно работать без фактической загрузки полного .csv в память: 

val blogsDF = sparkSession.read.schema(full_csv_schema)
                     .option("header", true)
                     .csv(file_path)
                     .select("Name","uniquie_number","state");
0 голосов
/ 27 апреля 2020

Вы пробовали это: 

schema = StructType([StructField("a", IntegerType(), True),
                    StructField("b", IntegerType(), True),
                    StructField("c", StringType(), True),
                    StructField("d", StringType(), True),
                    StructField("e", DoubleType(), True),
                    StructField("f", LongType(), True),
                   ])

df  = spark.read.csv('blablabla', schema=schema)
...