PySpark 2.4 - 使用自定义行分隔符读取 CSV 文件

Question

2017 年向 spark 添加了对自定义行分隔符（针对各种文本文件格式）的支持（参见：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 行数：3 two.csv 行数：3

我收到的输出是： one.csv 行数：3 two.csv 行数：1

关于如何让 Pyspark 接受组分隔符字符作为行分隔符的想法？

Answer 1

我可以用这个得到我想要的结果：

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()))

这取决于Pandas，这里的数据可能会被读取两次（我不确定当从 panda dataFrame 创建 RDD 时内部会发生什么）。

PySpark 2.4 - 使用自定义行分隔符读取 CSV 文件

PySpark 2.4 - Read CSV file with custom line separator

python

csv

text-parsing

apache-spark

pyspark