如何用PySpark解析EDI文件并拆分_c0列至单独行？

核心需求 #

用spark.read.csv加载EDI文件后，得到的_c0列为字符串类型，元素以单引号'作为分隔符，需要将每行中的每个元素拆分到单独行。

解决方案步骤 #

直接使用PySpark的split+explode组合即可实现，步骤如下：

1. 导入必要函数
   先导入PySpark SQL的函数库：

   from pyspark.sql import functions as F
   

2. 拆分字符串为数组
   使用split函数将_c0列按单引号拆分为数组，同时过滤掉拆分后产生的空字符串（避免EDI文件中多余的单引号导致无效行）：

   df_with_array = spark_df.withColumn(
       "elements", 
       F.filter(F.split(F.col("_c0"), "'"), lambda x: x != "")
   )
   

3. 展开数组为多行
   使用explode函数将数组中的每个元素拆分为单独的行，最后可以保留需要的列：

   final_df = df_with_array.select(F.explode(F.col("elements")).alias("split_element"))
   

完整代码示例 #

from pyspark.sql import functions as F

# 加载EDI文件（用户原加载逻辑）
spark_df = spark.read.csv(adls_path)

# 拆分并展开
df_with_array = spark_df.withColumn(
    "elements", 
    F.filter(F.split(F.col("_c0"), "'"), lambda x: x != "")
)
final_df = df_with_array.select(F.explode(F.col("elements")).alias("split_element"))

# 查看结果
final_df.show()

注意事项 #

• 如果EDI文件中的单引号存在转义（比如用两个单引号表示一个实际单引号），需要调整split的正则表达式，例如改为split(F.col("_c0"), "(?<!')'(?!')")来匹配作为分隔符的单引号，避免拆分元素内部的转义单引号。
• 若原数据行本身为空或无有效元素，filter会将数组置空，explode会自动过滤掉这些空数组对应的行，无需额外处理。

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