PySpark 2.4 - чтение CSV-файла с пользовательским разделителем строк - PullRequest
Новая
       12

PySpark 2.4 - чтение CSV-файла с пользовательским разделителем строк

0 голосов
/ 18 февраля 2019

Поддержка специальных разделителей строк (для различных форматов текстовых файлов) была добавлена ​​в Spark в 2017 году (см .: https://github.com/apache/spark/pull/18581).

... или, возможно, не была добавлена ​​в 2017 году - или когда-либо (см.: https://github.com/apache/spark/pull/18304)

Сегодня, с Pyspark 2.4.0, я не могу использовать пользовательские разделители строк для анализа файлов CSV.

Вот код: 

from pyspark.sql.types import (
    StructType,
    StructField,
    StringType
)

list_structs = StructType([StructField('id', StringType(), True),StructField('desc', StringType(), True)])

df = spark.read.load("mnt/one.csv",
                     format="csv", 
                     sep="\x1e",
                     schema=list_structs)
print("one.csv rowcount: {}".format(df.count()))

df2 = spark.read.load("mnt/two.csv",
                     format="csv", 
                     sep="\x1e",
                     lineSep="\x1d",
                     schema=list_structs)
print("two.csv rowcount: {}".format(df2.count()))

Вотдва примера файла CSV: one.csv - строки разделены символом перевода строки '0A' 

"1","foo"
"2","bar"
"3","foobar"

two.csv - строки разделены символом разделения группы '1D'

"1","foo""2","bar""3","foobar"

Я хочу, чтобы вывод из кода был следующим: one.csv rowcount: 3 two.csv rowcount: 3

Вывод, который я получаю: one.csvrowcount: 3 two.csv rowcount: 1

И идеи о том, как заставить Pyspark принять символ разделителя групп в качестве разделителя строк?

1 Ответ

0 голосов
/ 18 февраля 2019

Я могу получить желаемый результат с помощью этого: 

import pandas as pd

padf = pd.read_csv("/dbfs/mnt/two.csv",
                  engine="c",
                  sep="\x1e",
                  lineterminator ="\x1d",
                  header=None,
                  names=['id','desc'])
df = sqlContext.createDataFrame(padf)
print("two.csv rowcount: {}".format(df.count()))

Это зависит от Панд, и здесь данные могут быть прочитаны дважды (я не уверен, что происходит внутри, когда RDD создается изpanda dataFrame).

...