如何通过Azure Data Factory将特殊结构CSV导入PostgreSQL

问题根源 #

你遇到的重复双引号和json函数报错，本质是API导出的CSV对JSON字符串做了双重转义：JSON内部的双引号被转义为\"，同时CSV用双引号包裹每个JSON字段，导致ADF读取后字符串变成""{...}""的格式，不符合合法JSON的要求。

分步解决步骤 #

1. 调整CSV源的读取配置 #

• 在ADF的CSV数据集配置中，关闭第一行作为标题（因为首行是行号序列，不是实际字段名），或者保留并将其作为列名使用。
• 在数据流的源转换里，勾选跳过前1行，直接读取第二行的实际JSON数据行。

2. 清理JSON字符串的转义问题 #

对每个存储JSON的列（如Column0、Column1...），先清理掉首尾的冗余双引号，再用json函数解析：

• 派生列表达式（以Column0为例）：

  json(trim(Column0, '"'))
  

  • trim(Column0, '"')：去掉字符串首尾的双引号，得到合法的JSON文本。
  • json(...)：将清理后的文本解析为JSON对象，供后续提取字段使用。

3. 扁平化解嵌套与数组字段 #

从解析后的JSON对象中提取目标字段：

• 顶层字段直接提取：

  ParsedJSON.AccountID
  ParsedJSON.CarrierID
  

• 嵌套字段（如ShipToAddress的子字段）直接链式访问：

  ParsedJSON.ShipToAddress.Address1
  ParsedJSON.ShipToAddress.City
  

• 数组字段Column5去除括号：
  • 如果数组仅含单个元素，直接取第一个元素：

    ParsedJSON.Column5[0]
    

  • 如果数组含多个元素，转成字符串后移除首尾括号：

    replace(replace(string(ParsedJSON.Column5), '[', ''), ']', '')
    

4. 数据流完整流程 #

1. 源转换：连接到Azure Storage的CSV数据集，配置跳过首行。
2. 派生列转换：为每个JSON列生成清理后的JSON对象。
3. 二次派生列：从JSON对象中提取所有需要的扁平字段。
4. 选择转换：过滤掉中间过程的冗余列（如原始列、清理后的文本列）。
5. 接收器转换：连接到PostgreSQL数据集，映射字段后完成写入。

5. 报错排查 #

• 如果json函数仍报错，先在ADF预览中查看清理后的JSON文本，复制到本地JSON校验工具确认格式是否合法。
• 如果出现JSON被拆分的情况，检查CSV数据集的分隔符设置，确保API导出时每个JSON字段都被双引号包裹，避免内部逗号被误判为分隔符。

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