Azure Data Factory数据流因Null值无法读取JSON对象求解决方法

问题背景 #

ADF数据流读取JSON数组文件时，因数组内存在Null元素触发反序列化错误，报错信息如下：

Error occurred when deserializing source JSON file. Check if the data is in valid JSON object format. Error reading JObject from JsonReader. Current JsonReader item is not an object: Null. Path '[0]', line 2, position 5.

手动移除Null元素后数据流可正常运行，以下是几种自动化解决方法：

方法1：预处理文件（Copy Activity + Azure Function） #

通过ADF的Copy Activity将原文件复制到临时存储，再调用Azure Function过滤数组中的Null元素，处理后的文件作为数据流源：

1. 创建一个Azure Function，实现JSON数组过滤逻辑：

using System.IO;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.AspNetCore.Http;
using Newtonsoft.Json;
using System.Collections.Generic;

public static class FilterNullJsonItems
{
    [FunctionName("FilterNullJsonItems")]
    public static async Task<IActionResult> Run(
        [HttpTrigger(AuthorizationLevel.Function, "post", Route = null)] HttpRequest req,
        [Blob("processed-files/{filename}", FileAccess.ReadWrite)] Stream targetBlob)
    {
        // 读取请求中的JSON内容
        string rawJson = await new StreamReader(req.Body).ReadToEndAsync();
        dynamic jsonArray = JsonConvert.DeserializeObject(rawJson);
        
        // 过滤Null元素
        List<dynamic> filteredItems = new List<dynamic>();
        foreach (var item in jsonArray)
        {
            if (item != null) filteredItems.Add(item);
        }
        
        // 写入处理后的JSON到Blob
        string filteredJson = JsonConvert.SerializeObject(filteredItems);
        using (var writer = new StreamWriter(targetBlob))
        {
            await writer.WriteAsync(filteredJson);
        }
        
        return new OkObjectResult("Null elements filtered successfully");
    }
}

2. 在ADF中添加自定义活动，配置为调用上述Azure Function，传入原文件内容；
3. 将处理后的文件路径作为后续数据流的源路径。

方法2：数据流内直接过滤（需源可加载数组结构） #

如果ADF能识别数组结构（仅报错无法解析Null元素），可通过以下步骤处理：

• 步骤1：配置源：将源的JSON格式设置为Array of objects，勾选允许Schema漂移，关闭验证Schema；
• 步骤2：添加派生列：创建新列IsValid，表达式为isNotNull(item())；
• 步骤3：添加筛选转换：设置过滤条件为IsValid == true，仅保留非Null的数组元素；
• 步骤4：后续解析：对筛选后的元素进行正常的JSON解析或列映射。

方法3：使用数据流脚本直接过滤 #

直接修改数据流的脚本，在源之后添加过滤逻辑，跳过Null元素：

source(output(
    Body as array
),
allowSchemaDrift: true,
validateSchema: false) ~> RawSource
RawSource filter(isNotNull(Body)) ~> FilterNullElements
FilterNullElements flatten(Body, skipDuplicateMapInputs: true, skipDuplicateMapOutputs: true) ~> FlattenArray
// 后续添加解析或映射逻辑

内容的提问来源于stack exchange，提问作者Ravi Teja Potluri