ADF - 数据流 - 属性名称的 Json 表达式

如何解决ADF - 数据流 - 属性名称的 Json 表达式

我需要使用 Azure 数据工厂中的数据流将 json 转换为 csv（或 sql 表）或任何其他扁平化结构。我需要从源 json 中获取某些层次结构的属性名称和层次结构较低的子属性的值，并将它们添加为 csv 或任何其他扁平结构中的列/行值。

源数据规则/约束：

父级数据属性名称将动态更改（例如 ABCDataPoints、CementUse、CoalUse、ABCUseIndicators 名称是动态的）
层次结构始终与以下示例 json 中的相同。

我需要一些帮助来定义 Json 路径/表达式以获取名称 ABCDataPoints、CementUse、CoalUse、ABCUseIndicators 等。我能够弄清楚如何检索属性的值 Value,ValueDate、Valuescore、AsReported。

源数据结构：

{
"ABCDataPoints": {
    "CementUse": {
        "Value": null,"ValueDate": null,"Valuescore": null,"AsReported": [],"Sources": []
    },"CoalUse": {
        "Value": null,"Sources": []
    }
},"ABCUseIndicators": {
    "EnvironmentalControversies": {
        "Value": false,"ValueDate": "2021-03-06T23:22:49.870Z"
    },"RenewableEnergyUseRatio": {
        "Value": null,"Valuescore": null
    }
},"XYZDataPoints": {
    "AccountingControversiesCount": {
        "Value": null,"AdvanceNotices": {
        "Value": null,"Sources": []
    }        
},"XYXIndicators": {
    "AccountingControversies": {
        "Value": false,"AntiTakeoverDevicesAboveTwo": {
        "Value": 4,"ValueDate": "2021-03-06T23:22:49.870Z","Valuescore": "0.8351945854483925"
    }     
}

}

预期的扁平化结构

解决方法

背景：在与 Microsoft 的 ADF 专家多次通话后（我们的工作场所与 Microsoft/Azure 合作），他们得出的结论是，ADF 提供的开箱即用活动是不可能的，Dataflow 也不可能（不需要）使用数据流）也没有展平功能。原因是 Dataflow/Flatten 仅展开 Array 对象，并且没有可用于选择属性名称的映射函数 - 自定义表达式正在内部 beta 测试中，并将在不久的将来在 PA 中使用。

结论/解决方案：我们基于与 Microsoft emps 的调用达成了协议，最终采用多种方法，但两者都需要自定义代码 - 如果没有自定义代码，则无法使用开箱即用的活动。

解决方案 1：使用 ADF 自定义活动，根据要求使用一些代码进行展平。这样做的缺点是您需要使用外部计算（VM/Batch），支持的选项不是按需提供的。所以它有点贵，但如果有连续的流工作负载，效果最好。这种方法还会持续监控输入源是否具有不同大小，因为在这种情况下计算需要具有弹性，否则您将遇到内存不足异常。

解决方案 2：仍然需要编写自定义代码 - 但在函数应用中。

创建一个复制活动，源文件是包含 Json 内容的文件（最好是存储帐户）。
使用目标作为函数的休息端点（不作为函数活动，因为从 ADF 活动调用时它有 90 秒超时）
函数应用将接受 Json 行作为输入并解析和展平。
如果您使用上述方式，那么您可以扩展每个请求中发送的行数以使其正常工作，并扩展并行请求。
该函数将根据需要对一个或多个文件进行展平并存储在 blob 存储中。
管道将根据需要从那里继续。
这种方法的一个问题是，如果任何范围失败，复制活动将重试，但它会再次运行整个过程。