如何使用pyspark读取带有引号中附加逗号的csv文件？

Question

在UTF-16中读取以下CSV数据时遇到一些问题：

FullName, FullLabel, Type
TEST.slice, "Consideration":"Verde (Spar Verde, Fonte Verde)", Test,

据我所知，这对读者来说应该不是问题，因为有一个quote参数来处理这个问题

df = spark.read.csv(file_path, header=True, encoding='UTF-16', quote = '"')

但是，这仍然会给我一个不正确的分割：

有没有办法处理这些情况，或者我需要与RDD一起解决

先谢谢你

Answer 1

您可以使用spark.read.text作为文本读取，并使用一些正则表达式拆分值，以逗号分隔，但忽略引号（您可以看到这个post），然后从结果数组中获取相应的列：

from pyspark.sql import functions as F

df = spark.read.text(file_path)

df = df.filter("value != 'FullName, FullLabel, Type'") \
    .withColumn(
    "value",
    F.split(F.col("value"), ',(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)')
).select(
    F.col("value")[0].alias("FullName"),
    F.col("value")[1].alias("FullLabel"),
    F.col("value")[2].alias("Type")
)

df.show(truncate=False)

#+     +                         +  -+
#|FullName  |FullLabel                                         |Type |
#+     +                         +  -+
#|TEST.slice| "Consideration":"Verde (Spar Verde, Fonte Verde)"| Test|
#+     +                         +  -+

更新：

对于utf-16中的输入文件，可以通过将文件加载为binaryFiles来替换spark.read.text，然后将结果rdd转换为数据帧：

df = sc.binaryFiles(file_path) \
    .flatMap(lambda x: [[l] for l in x[1].decode("utf-16").split("\n")]) \
    .toDF(["value"])

Answer 2

下面是另一个选项（如果您觉得简单）：

首先将文本文件读取为RDD，并将":"替换为~:~并保存文本文件

sc.textFile(file_path).map(lambda x: x.replace('":"','~:~')).saveAsTextFile(tempPath)

接下来，读取临时路径并再次将~:~替换为":"，但这次是作为DF

from pyspark.sql import functions as F
spark.read.option('header','true').csv(tempPath).withColumn('FullLabel',F.regexp_replace(F.col('FullLabel'),'~:~','":"')).show(1, False)

+     +                       -+  +
|FullName  |FullLabel                                      |Type|
+     +                       -+  +
|TEST.slice|Consideration":"Verde (Spar Verde, Fonte Verde)|Test|
+     +                       -+  +