Azure Data Factory处理列数可变CSV转Parquet的方案咨询

我之前处理过不少这种混合结构的CSV文件，ADF默认的CSV解析确实会因为第一行列数限制而丢数据，给你一套数据流的解决方案，亲测有效：

核心思路 #

因为你的文件包含**头（3列）、明细（5列）、尾（2列）**三种不同列数的行，不能直接用标准CSV解析。我们需要先把每行当成纯文本读取，再通过判断行标识拆分不同类型的记录，最后单独处理明细数据转Parquet。

步骤1：数据流源配置（关键！） #

在数据流的源中，将CSV的列分隔符设置为一个绝对不会出现在你的数据中的特殊字符（比如\u0000，也就是空字符）。这样ADF会把整个文件的每行都读取为一个单独的字符串列（默认命名为rawLine），避免因为第一行列数限制截断后续列。

步骤2：派生记录类型列 #

添加一个派生列转换，通过行首字符判断当前行属于头、明细还是尾记录：

recordType = left(rawLine, 1)

对应你的示例数据：

• 头记录行首是1
• 明细记录行首是2
• 尾记录行首是99（这里注意如果尾记录是两位开头，要改判断逻辑为left(rawLine, 2) == '99'）

步骤3：条件拆分分离不同记录 #

添加条件拆分转换，设置三个分支：

• 头记录分支：recordType == '1'
• 明细记录分支：recordType == '2'
• 尾记录分支：left(rawLine, 2) == '99'

步骤4：处理明细记录 #

针对明细记录分支，添加拆分列转换：

1. 选择rawLine作为拆分列
2. 拆分类型选自定义，输入你的分隔符|~
3. 因为明细是5列，拆分后会生成5个新字段（比如col1到col5），对应示例中的2、Mark、cse、378、2020

步骤5：输出到Parquet #

将处理好的明细字段映射到Parquet目标的对应列，然后运行数据流即可。如果需要保留头/尾记录，可以将对应分支的数据单独输出到其他存储，按需处理。

备选方案：用Azure Databricks处理 #

如果数据流的配置觉得麻烦，也可以用Databricks的Spark代码来处理，灵活性更高：

val df = spark.read.text("abfss://<container>@<account>.dfs.core.windows.net/<file-path>")

// 拆分不同类型记录
val headerDF = df.filter($"value".startsWith("1|~")).select(split($"value", "\\|~").alias("cols"))
val detailDF = df.filter($"value".startsWith("2|~")).select(
  split($"value", "\\|~")(0).alias("recordFlag"),
  split($"value", "\\|~")(1).alias("name"),
  split($"value", "\\|~")(2).alias("dept"),
  split($"value", "\\|~")(3).alias("id"),
  split($"value", "\\|~")(4).alias("year")
)
val footerDF = df.filter($"value".startsWith("99|~")).select(split($"value", "\\|~").alias("cols"))

// 将明细写入Parquet
detailDF.write.parquet("abfss://<container>@<account>.dfs.core.windows.net/<output-path>")

这样就能完美解决列数不匹配的问题，不会再丢失明细的额外列了。

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