使用 pyspark 读取 csv 文件时获取格式错误记录的列名答案

【问题标题】：Get the column names of malformed records while reading a csv file using pyspark使用 pyspark 读取 csv 文件时获取格式错误记录的列名
【发布时间】：2020-10-01 09:33:17
【问题描述】：

我正在使用预定义架构的 pyspark 读取 csv 文件。

schema = StructType([
StructField("col1", IntegerType(), True),
StructField("col2", StringType(), True)
StructField("col3", FloatType(), True)
])

df = spark.sqlContext.read
    .schema(schema)
    .option("header",true)
    .option("delimiter", ",")
    .csv(path)

现在在 csv 文件中，col1 中有浮点值，col3 中有字符串值。我需要引发异常并获取这些列的名称（col1，col3），因为这些列包含的数据类型与架构中定义的数据类型不同。

我如何做到这一点？

【问题讨论】：

标签： scala csv apache-spark pyspark apache-spark-sql

【解决方案1】：

在 pyspark 版本 >2.2 中，您可以将 columnNameOfCorruptRecord 与 csv 一起使用：

schema = StructType(
    [
        StructField("col1", IntegerType(), True),
        StructField("col2", StringType(), True),
        StructField("col3", FloatType(), True),
        StructField("corrupted", StringType(), True),
    ]
)

df = spark.sqlContext.read.csv(
    path,
    schema=schema,
    header=True,
    sep=",",
    mode="PERMISSIVE",
    columnNameOfCorruptRecord="corrupted",
).show()

+----+----+----+------------+
|col1|col2|col3|   corrupted|
+----+----+----+------------+
|null|null|null|0.10,123,abc|
+----+----+----+------------+

编辑： CSV 记录字段不是相互独立的，因此通常不能说一个字段已损坏，而其他字段则没有。只有整个记录可以损坏或不损坏。

例如，假设我们有一个逗号分隔的文件，其中包含一行和两列浮点数，欧元值 0,10 和 1,00。该文件如下所示：

col1,col2
0,10,1,00

哪个字段损坏了？

【讨论】：

这会给出整行格式错误的值。我需要像 col1 和 col3 这样的列名。我如何获得这些列名？
好问题。见我上面的回答。
谢谢戴夫。在欧洲编号系统的情况下确实如此。在 python 中，我们可以指定参数 decimal= ',' 。但在 pyspark 我想没有这样的选择。唯一的方法是将数据读取为字符串，然后使用正则表达式。但就我而言，我需要知道列名。有没有办法得到它？