Synapse元数据驱动管道处理含管道符的CSV文件报错求助

问题场景 #

通过Synapse管道的存储过程（SP）活动，将Blob存储中以竖线|为分隔符的CSV文件加载至临时表，采用元数据驱动架构。执行时触发报错：

Code":21104,"Message":"ErrorCode=DelimitedTextMoreColumnsThanDefined,'Type=Microsoft.DataTransfer.Common.Shared.HybridDeliveryException,Message=处理'Csv/Tsv Format Text'源文件'PomOP_PRF_Status20240311_033449-00001.txt'的第11行时出错：发现的列数超出预期的14列。,Source=Microsoft.DataTransfer.Common,'

经排查，问题源于源数据某列值内部包含分隔符|，示例行：
2307906|1/17 - Bridge product may need to be authorized for next infusion1st Infusion 11/30/2023 | PA Approved 11/1/2023

其中1/17 - Bridge product may need to be authorized for next infusion1st Infusion 11/30/2023 | PA Approved 11/1/2023应为单个列值，但内部竖线被误判为分隔符，导致列数超出预期，且无法修改源文件。

可行解决方案 #

方案1：利用文本限定符或数据流动预处理 #

• 如果源文件中包含内部分隔符的列值有统一的文本限定符（比如双引号），直接在元数据表的CSV格式配置中添加QuoteCharacter参数，值设为对应限定符（如"）。Synapse会自动将限定符包裹的内容识别为单个列值，忽略内部的分隔符。
• 若源文件无限定符，用数据流动预处理：
  1. 管道中添加数据流动活动，源连接到Blob存储的目标CSV文件。
  2. 源转换的Source Options中，不指定分隔符，将整行作为单个字符串列读取。
  3. 添加派生列转换，用表达式合并被误拆分的列。针对示例行（预期14列，第2列含内部竖线），表达式示例：

     concat(split($$,'|')[0], '|', join(slice(split($$,'|'),1,13), '|'))
     

  4. 将处理后的行写入临时Blob路径，再通过SP活动加载到临时表。

方案2：修改存储过程的加载逻辑 #

• 改用OPENROWSET或BULK INSERT配合自定义格式文件：
  1. 创建XML格式文件，精准定义每列的位置、分隔符。针对含内部竖线的列，指定其结束位置为下一个正确的分隔符（比如第3个|之前的内容作为第2列）。
  2. 存储过程中调用OPENROWSET时引用该格式文件，示例代码：

     SELECT * INTO #TempTable
     FROM OPENROWSET(
         BULK 'https://yourstorage.blob.core.windows.net/container/PomOP_PRF_Status20240311_033449-00001.txt',
         FORMATFILE = 'https://yourstorage.blob.core.windows.net/container/format_file.xml',
         FIRSTROW = 2
     ) AS data;
     

  注意：需要提前分析源文件的列结构，确保格式文件能准确识别目标列。

方案3：自定义活动（Azure Function）预处理 #

• 如果数据流动表达式处理复杂，添加自定义活动调用Azure Function，逐行处理源文件：
  1. 编写代码读取源文件每行内容，判断列数是否超出预期。
  2. 将包含内部竖线的列值合并，或用限定符包裹后写入新文件。
  3. 后续SP活动加载处理后的新文件。
     示例C#代码片段：

     string line = await reader.ReadLineAsync();
     string[] parts = line.Split('|');
     if (parts.Length > 14)
     {
         // 合并第2到倒数第13个元素，恢复为单个列
         string mergedColumn = string.Join("|", parts.Skip(1).Take(parts.Length - 13));
         List<string> newParts = new List<string> { parts[0], mergedColumn };
         newParts.AddRange(parts.Skip(parts.Length - 12));
         line = string.Join("|", newParts);
     }
     await writer.WriteLineAsync(line);
     

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