PySpark读取CSV时如何忽略双引号并提取ApplicationID？

我有如下格式的CSV文件：

ID, Tags
1,"""ApplicationID """: """9AAG033396""","""Environment """: """PROD""","""Remarks""": """ EUC Personal Desktop"""
2,"""ApplicationID """: """9AAG033396""","""Environment """: """PROD""","""Remarks""": """ EUC Personal Desktop"""
3,"""Remarks""": """ EUC Personal Desktop""","""ApplicationID """: """9AAG033396""","""Environment """: """PROD"""

Tags字段包含多个带嵌套双引号的键值对。我想用PySpark将其读取为DataFrame，要求：

1. 去除Tags字段中所有双引号
2. 新增ApplicationID列，提取对应的值

预期输出如下：

ID, Tags,ApplicationID
1,ApplicationID : 9AAG033396,Environment : PROD,Remarks:  EUC Personal Desktop,9AAG033396
2,ApplicationID : 9AAG033396,Environment : PROD,Remarks:  EUC Personal Desktop,9AAG033396
3,Remarks:  EUC Personal Desktop,ApplicationID : 9AAG033396,Environment : PROD,9AAG033396

但当前读取后Tags字段残留双引号，且部分ApplicationID值为空，我的PySpark代码如下：

df = spark.read\
  .option("header", "true")\
  .option("inferSchema", "true")\
  .option("delimiter", ",")\
  .option("escapeQuotes", "true")\
  .option("multiLine","true")\
  .option('quote','"')\
.csv(f"wasbs://{container}@{storage_account_name}.blob.core.windows.net/onetimeazamortizecostnew/onetimeazamortizecostnew/20230901-20231031/onetimeazamortizecostnew_17fcdeca-81ca-43e7-b181-36bc379e9644.csv")

df2 = df.withColumn("ApplicationID", when(df.Tags.contains("ApplicationID"),substring('Tags', 23,10))
                                    .otherwise(''))

请问能否通过设置escape选项来实现需求？

可以通过设置escape选项配合后续字段处理来实现需求，具体步骤如下：

1. 调整CSV读取参数 #

你的CSV中用三个双引号包裹内容，内部的双引号是转义后的格式，需要设置escape="\""（即转义字符为双引号），同时保留quote="\""，这样Spark能正确解析嵌套的双引号。修改后的读取代码：

df = spark.read\
  .option("header", "true")\
  .option("inferSchema", "true")\
  .option("delimiter", ",")\
  .option("multiLine", "true")\
  .option("quote", '"')\
  .option("escape", '"')\  # 关键：指定转义字符为双引号
.csv(f"wasbs://{container}@{storage_account_name}.blob.core.windows.net/onetimeazamortizecostnew/onetimeazamortizecostnew/20230901-20231031/onetimeazamortizecostnew_17fcdeca-81ca-43e7-b181-36bc379e9644.csv")

设置escape="\""后，Spark会把""解析为单个"，此时Tags字段内容会变成带单个双引号的字符串，比如"ApplicationID ": "9AAG033396","Environment ": "PROD","Remarks": " EUC Personal Desktop"。

2. 去除Tags字段所有双引号 #

用regexp_replace函数替换掉所有双引号：

from pyspark.sql.functions import regexp_replace, regexp_extract

df_clean = df.withColumn("Tags", regexp_replace("Tags", '"', ""))

处理后Tags字段就会变成预期的无引号格式：ApplicationID : 9AAG033396,Environment : PROD,Remarks: EUC Personal Desktop。

3. 可靠提取ApplicationID #

之前用substring的方式不可靠，因为ApplicationID的位置可能变化（比如第3行的位置就不一样），改用regexp_extract来匹配键值对：

df_final = df_clean.withColumn(
    "ApplicationID",
    regexp_extract("Tags", r"ApplicationID\s*:\s*(\w+)", 1)
)

这个正则表达式会匹配ApplicationID后面的冒号和空格，提取对应的ID值，不管它在Tags中的位置如何。

完整代码 #

from pyspark.sql.functions import regexp_replace, regexp_extract

# 读取CSV，设置正确的转义参数
df = spark.read\
  .option("header", "true")\
  .option("inferSchema", "true")\
  .option("delimiter", ",")\
  .option("multiLine", "true")\
  .option("quote", '"')\
  .option("escape", '"')\
.csv(f"wasbs://{container}@{storage_account_name}.blob.core.windows.net/onetimeazamortizecostnew/onetimeazamortizecostnew/20230901-20231031/onetimeazamortizecostnew_17fcdeca-81ca-43e7-b181-36bc379e9644.csv")

# 清理Tags字段的双引号
df_clean = df.withColumn("Tags", regexp_replace("Tags", '"', ""))

# 提取ApplicationID
df_final = df_clean.withColumn(
    "ApplicationID",
    regexp_extract("Tags", r"ApplicationID\s*:\s*(\w+)", 1)
)

# 查看结果
df_final.show(truncate=False)

验证结果 #

执行后会得到符合预期的输出：

+---+---------------------------------------------------------------------+-------------+
|ID |Tags                                                                 |ApplicationID|
+---+---------------------------------------------------------------------+-------------+
|1  |ApplicationID : 9AAG033396,Environment : PROD,Remarks:  EUC Personal Desktop|9AAG033396   |
|2  |ApplicationID : 9AAG033396,Environment : PROD,Remarks:  EUC Personal Desktop|9AAG033396   |
|3  |Remarks:  EUC Personal Desktop,ApplicationID : 9AAG033396,Environment : PROD|9AAG033396   |
+---+---------------------------------------------------------------------+-------------+

内容的提问来源于stack exchange，提问作者Deepak Kumar