ADF-数据流-属性名称的JSON表达式 [英] ADF - Data Flow- Json Expression for Property name

问题描述

我需要使用Azure Data Factory中的数据流将json转换为CSV(或SQL表)或任何其他扁平化结构。我需要从源JSON中获取某些层次结构中的属性名称和层次结构较低的子属性的值，并将它们作为列/行值添加到CSV或任何其他扁平化结构中。

源数据规则/约束：

1. 父级数据属性名称将动态更改(例如，ABCDataPoints、CementUse、CoalUse、ABCUseIndicator名称是动态的)
2. 层次结构始终与下面的示例json相同。

我需要一些帮助来定义Json路径/表达式以获取名称ABCDataPoints、CementUse、CoalUse、ABCUseIndicator等。我能够确定如何检索Value、ValueDate、ValueScore、AsReated属性的值。

源数据结构：

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

}

预期的扁平化结构

推荐答案

背景： 在与微软的ADF专家多次通话后(我们的工作场所有Microsoft/Azure合作伙伴)，他们得出结论，这不可能通过ADF按原样提供的开箱即用活动实现，也不可能通过数据流(尽管不需要使用数据流)或扁平化功能来实现。原因是数据流/扁平化仅展开数组对象，并且没有可用于选取属性名称的映射函数-自定义表达式正在进行内部Beta测试，不久将在PA中。

结论/解决方案： 我们最终与Microsoft EMP达成了一项基于呼叫的协议，最终采用了多种方法，但两者都需要定制代码-如果没有定制代码，使用开箱即用的活动是不可能的。

解决方案-1：使用ADF Custom Activity根据要求使用一些代码进行扁平化。缺点是您需要使用外部计算机(虚拟机/批处理)，支持的选项不是按需提供的。所以它有点贵，但如果有连续的流工作负载，效果最好。此方法还会持续监视输入源是否具有不同的大小，因为在这种情况下，计算需要具有弹性，否则会出现内存不足异常。

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

• 创建一个复制活动，将源作为包含Json内容的文件(最好是存储帐户)。
• 将目标用作函数的REST终结点(不是函数活动，因为它在从ADF活动调用时有90秒超时)
• 函数应用程序将接受Json行作为输入，并进行解析和拼合。
• 如果您使用上述方法，则可以扩展每个请求中发送的行数以发挥作用，还可以扩展并行请求。
• 该函数将根据需要对一个或多个文件进行展平，并存储在Blob存储中。
• 管道将根据需要从那里继续。
• 此方法的一个问题是，如果任何范围失败，复制活动将重试，但它将再次运行整个过程。

这篇关于ADF-数据流-属性名称的JSON表达式的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！