Azure Data Factory中API转Blob存储时JSON字段结构多变致schema识别异常

一、用数据流派生列处理字段变体 #

针对answered_by和from的结构差异，直接在数据流中通过表达式统一格式：

处理answered_by字段 #

answered_by存在空数组、含contact对象的数组、含number字段的数组三种情况，用以下表达式提取目标值（示例取第一个有效元素的对应值）：

iif(arrayLength(answered_by) == 0, 
    null(), 
    iif(exists(answered_by, item => item.contact != null), 
        answered_by[0].contact.name,  // 按需替换为你需要的contact字段，比如id/number
        answered_by[0].number
    )
)

如果数组包含多个元素，可改用map函数批量处理，或用aggregate合并结果。

处理from字段 #

from可能是字符串或contact对象，用typeof判断类型后提取值：

iif(typeof(from) == 'string', 
    from, 
    from.contact.number  // 按需替换为你需要的contact字段
)

二、启用Schema Drift配合动态映射 #

如果字段变体更复杂，可在数据源中开启Schema Drift，允许读取未定义的列，再通过动态表达式兼容所有结构：

1. 在API源数据集的Schema设置中勾选「Allow schema drift」
2. 在派生列或映射环节，用coalesce函数依次尝试读取不同结构的字段：
   比如处理from的简化写法：

   coalesce(from.number, from.contact.number, from)
   

   该表达式会优先读取from.number，不存在则读from.contact.number，最后直接取字符串类型的from。

三、Azure Function预处理JSON（复杂场景） #

如果数据流表达式无法满足需求，可先用Lookup活动读取API返回的JSON，再调用Azure Function转换结构：
用Python编写Function逻辑示例：

def main(req):
    input_json = req.get_json()
    
    # 标准化answered_by
    ab_items = input_json.get('answered_by', [])
    input_json['answered_by_normalized'] = None
    if ab_items:
        first_item = ab_items[0]
        if 'contact' in first_item:
            input_json['answered_by_normalized'] = first_item['contact']['name']  # 按需调整字段
        elif 'number' in first_item:
            input_json['answered_by_normalized'] = first_item['number']
    
    # 标准化from
    from_field = input_json.get('from')
    if isinstance(from_field, str):
        input_json['from_normalized'] = from_field
    else:
        input_json['from_normalized'] = from_field.get('contact', {}).get('number')  # 按需调整字段
    
    return input_json

处理完成后，将转换后的JSON写入Blob存储即可。

关键注意点 #

• 测试时必须覆盖所有结构场景（空数组、contact对象、number对象/字符串），验证表达式逻辑
• 若数组含多个元素，需明确处理规则（取第一个、合并所有值等）
• 开启Schema Drift后注意监控性能，大体积JSON建议分批处理

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