SPARK: файлы читаются в каталоге, в некоторых файлах отсутствуют столбцы с заголовком столбца

Question

У меня есть 4 файла в каталоге, и одному файлу не хватает одного столбца с данными этого столбца

Но когда я загружаюсь в DF-искру, он не добавляет 2-й столбец и не заполняет ноль

                    file1.csv
                    name| first|second|
                    female|   raj| tarun|

                    file2.csv
                    name| first|second|
                    female|   raj| tarun|

                    file3.csv
                    name| first|second|
                    female|   raj| tarun|


                    file4.csv
                    name| second|
                    female|  tarun|





                    from pyspark.sql import SQLContext
                    sqlContext = SQLContext(sc)
                    from pyspark import SparkConf, SparkContext



                    un = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').option("delimiter",",").load('/dir/test/')
                    un.show()
                    un.registerTempTable("un1")


                    queryresult1 = sqlContext.sql("select DISTINCT hashedId from un1   ")


                    queryresult1.show()


                    Output is :  why second column is not filling with nul and third column did not shifted

                    +------+------+------+
                    |  name| first|second|
                    +------+------+------+
                    |female|   raj| tarun|
                    |female|   raj| tarun|
                    |female|   raj| tarun|
                    |  name|second|  null|
                    |female| tarun|  null|

У меня есть 4 файла в каталоге, и одному файлу не хватает одного столбца с данными этого столбца

Но когда я загружаюсь в DF-искру, он не добавляет 2-й столбец и не заполняет ноль

Answer 1

Не проверили это полностью, но приведенный ниже код должен помочь вам начать:

columns=['name', 'first', 'second']

df = sc.textFile(path to your folder)
    .map(lambda line: line.split("|")
    .filter(lambda line: line[0]!='name')
    .map(lambda line: line if len(line)==3 else [line[0],None, line[1]])
    .toDF(schema=columns)

Объяснение: Считать как СДР и разделить на разделитель труб. Фильтр, чтобы избавиться от строки заголовка в каждом разделе. Затем, если столбец отсутствует (т. Е. Длина элемента rdd равна 2), добавьте ноль. Наконец, преобразуйте в фрейм данных.