azure - 如何将数据从 Apache Avro 提取到 Azure 数据资源管理器中？

Question

这几天我一直在尝试将 Apache Avro 格式的数据从 Blob 存储提取到 Azure 数据资源管理器中。

我可以引用顶级 JSON 键，例如 $.Body (请参阅下面屏幕截图中红色下划线的示例)，但是当它转到嵌套 JSON 键时，Azure 无法正确解析它们并且不显示任何内容(如绿色栏中所示:我希望 $.Body.entityId 引用 Body-JSON 中的键“entityId”)。

非常感谢您的帮助!

Here is a screenshot of the azure data explorer web interface

编辑 1

我已经尝试将“嵌套级别”选项增加到 2，但我得到的只是 is this error message with no further details 。当我将级别降低回 1 时，错误消息甚至不会消失。我必须取消并开始整个过程​​。

我刚刚意识到自动生成的列有一些奇怪的类型。看起来它们加起来就是 string 类型...这对我来说也有点奇怪。

编辑2

这是一些 kql 代码。

这是我的输入 .avro 文件的架构，是我从 Eventhub-Capture 中获取的:

{
  SequenceNumber: ...,
  Offset: ...,
  EnqueuedTimeUTC: ...,
  SystemProperties: ...,
  Properties: ...,
  Body: {
    entityId: ...,
    eventTime: ...,
    messageId: ...,
    data: ...
  }
}, ...

通过这些摄取命令，我无法引用内部 JSON key 。顶级按键工作得很好。

// Create table command
////////////////////////////////////////////////////////////
.create table ['test_table']  (['Body']:dynamic, ['entityId']:string)

// Create mapping command
////////////////////////////////////////////////////////////
.create table ['test_table'] ingestion apacheavro mapping 'test_table_mapping' '[{"column":"Body", "Properties":{"Path":"$.Body"}},{"column":"entityId", "Properties":{"Path":"$.Body.entityId"}}]'

// Ingest data into table command
///////////////////////////////////////////////////////////
.ingest async into table ['test_table'] (h'[SAS URL]') with (format='apacheavro',ingestionMappingReference='test_table_mapping',ingestionMappingType='apacheavro',tags="['503a2cfb-5b81-4c07-8658-639009870862']")

我希望提取单独列上的内部数据字段，而不是使用更新策略构建任何解决方法。

Answer 1

对于那些遇到同样问题的人，以下是我们当前使用的解决方法:

首先，假设我们想要将 Body 字段的内容从 avro 文件提取到表 avro_destination。

第 1 步:创建摄取表

.create table avro_ingest(
    Body: dynamic
    // optional other columns, if you want...
)

第 2 步:创建更新政策

.create-or-alter function
    with (docstring = 'Convert avro_ingest to avro_destination', folder='ingest')
    convert_avro_ingest() {
        avro_ingest
        | extend entityId = tostring(Body.entityId)
        | extend messageId = tostring(Body.messageId)
        | extend eventTime = todatetime(Body.eventTime)
        | extend data = Body.data
        | project entityId, messageId, eventTime, data
    }

.alter table avro_destination policy update
@'[{ "IsEnabled": true, "Source": "avro_ingest", "Query": "convert_avro_ingest()", "IsTransactional": false, "PropagateIngestionProperties": true}]'

第 3 步:将 .avro 文件提取到 avro_ingest 表中

...如问题中所示，每个条目有一列包含整个 Body-JSON。