如何用pyspark读取引号中带有额外逗号的csv文件？

1 人关注

我在 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来解决这个问题？
预先感谢你。


           
            
             试着像这样引用： quote = '\"' 。这就是使用转义字符。让我们知道这是否有效。


           
            
             引号应该包括整个栏目，而不是栏目的部分内容


           
            
             @vijayinani 很遗憾没有。


           
            
             在这里你将需要一些词组。一些重词专家可以提供帮助。


         
          python


         
          apache-spark


         
          pyspark


         
          apache-spark-sql


         
          pyspark-dataframes


          
           
            
            
             blackbishop
            
           
           
            发布于
            
            2021-02-07


          
           已采纳


          
           
            你可以使用
            
             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中的输入文件，你可以将spark.read.text替换为binaryFiles，然后将生成的rdd转换成数据帧。
df = sc.binaryFiles(file_path) \
    .flatMap(lambda x: [[l] for l in x[1].decode("utf-16").split("\n")]) \
    .toDF(["value"])


          
           
            
             
              只是另一种选择，如下所示（如果你觉得简单的话）。
             
             
              首先将文本文件读为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)
+----------+-----------------------------------------------+----+