Azure Data Factory中如何将含\r\n的CSV格式字符串转为JSON数组

处理位置选择 #

推荐两种常用场景：

• 数据量较大或需复杂ETL逻辑：用**数据流（Data Flow）**处理
• 数据量小、逻辑简单：直接在Pipeline管道中用变量+表达式处理

方法一：用数据流转成JSON数组 #

步骤1：导入CSV字符串到数据流 #

通过Source组件或Derived Column生成包含目标CSV字符串的列（比如命名为RawCSV）

步骤2：拆分CSV为行数组 #

在Derived Column中添加计算列，拆分并过滤空行：

filter(split(RawCSV, '\r\n'), item() != '')

将该列命名为FilteredRows

步骤3：提取表头与数据行 #

• 表头列：split(first(FilteredRows), ',')（命名为Headers）
• 数据行数组：slice(FilteredRows, 1)（命名为DataRows）

步骤4：拆分数据行为字段 #

用Flatten组件将DataRows数组拆分为单行数据，再添加Derived Column拆分每行的字段：

split(currentRow, ',')

命名为Fields，同时添加过滤条件只保留字段数与表头一致的行：

equals(length(Fields), length(Headers))

步骤5：映射字段为JSON对象 #

用Derived Column生成JSON对象：

createObject(zip(Headers, Fields))

命名为JsonItem

步骤6：生成JSON数组 #

用Aggregate组件，将所有JsonItem收集为数组：

collect(JsonItem)

最终输出该数组即可得到目标格式的JSON

方法二：用Pipeline管道表达式处理 #

步骤1：定义变量 #

在Pipeline中创建以下变量：

• varRawCSV：存储原始CSV字符串
• varRows：存储拆分后的行数组
• varHeaders：存储表头数组
• varDataRows：存储数据行数组
• varResultArray：存储最终JSON数组（初始值设为[]）

步骤2：拆分并处理行数组 #

在Set Variable活动中设置varRows：

@filter(split(variables('varRawCSV'), '\r\n'), item() != '')

设置varHeaders：

@split(first(variables('varRows')), ',')

设置varDataRows：

@slice(variables('varRows'), 1)

步骤3：遍历数据行生成JSON对象 #

添加ForEach活动，遍历varDataRows，在活动内添加Set Variable：

• 先判断当前行字段数是否与表头一致：

  @if(equals(length(split(item(), ',')), length(variables('varHeaders'))), split(item(), ','), null)
  

• 若字段数匹配，生成JSON对象并添加到varResultArray：

  @concat(variables('varResultArray'), if(empty(variables('varResultArray')), '', ','), string(createObject(zip(variables('varHeaders'), split(item(), ',')))))
  

步骤4：生成最终JSON数组 #

最后用Set Variable包裹数组格式：

@concat('[', variables('varResultArray'), ']')

注意事项 #

• 原字符串中的\r\n在ADF表达式中直接用'\r\n'即可识别，无需额外转义
• 示例中第二行存在字段数不一致的问题（表头5列，该行6列），需通过过滤条件排除错误行，避免生成无效JSON
• 若需要保留错误行单独处理，可在过滤逻辑中分流存储

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