在流分析查询中将已知属性从记录提升到***结果
Posted
技术标签:
【中文标题】在流分析查询中将已知属性从记录提升到***结果【英文标题】:Promoting known properties from record to top level result in Stream Analytics Query 【发布时间】:2017-10-15 16:03:41 【问题描述】:我有一系列动态数据正在由流分析作业处理。我可以显式查询一些统一属性,但大部分有效负载在查询时属于未知类型。我的目标是获取这些未知数据(记录)并将所有属性提升到写入 Azure 表的结果查询中的***字段。
我能够展平记录的属性,它总是作为子对象添加到查询中。 GetRecordProperties()
没有帮助,因为我不希望为每个属性返回单独的记录。
我的查询如下所示:
WITH
[custom_events_temp] AS
(
SELECT
[magellan].[context].[data].[eventTime] as [event_time],
[flat_event].ArrayValue.name as [event_name],
udf.FlattenCustomDimensions([magellan].[context].[custom].[dimensions]) as [flat_custom_dim]
FROM [Magellan--AI-CustomEvents] magellan
TIMESTAMP BY [magellan].[context].[data].[eventTime]
CROSS APPLY GetElements([magellan].[event]) as [flat_event]
),
-- create table with extracted webhook data
[all_webhooks] AS
(
SELECT
[flat_custom_dim].[hook_event_source] as PartitionKey,
udf.CreateGuid('') as RowKey,
-- event data
[custom_events_temp].[event_time],
[custom_events_temp].[flat_custom_dim].[hook_event_name] as [event_name],
-- webhook payload data
udf.FlattenWebhookPayload(udf.ExtractJsonWebhookPayload([custom_events_temp].[flat_custom_dim].[webhook_payload])) AS [payload]
FROM [custom_events_temp]
)
SELECT * INTO [TrashTableOut] FROM [all_webhooks]
我得到的结果记录是这样的。这个想法是让 payload
嵌套对象中的所有内容不被嵌套,因此每个属性在 Azure 表中都有自己的列。
"partitionkey": "zzzzzzzzz",
"rowkey": "8beeb783-b07f-8a98-ef56-71c43378a5fc",
"event_time": "2017-10-15T05:37:06.3240000Z",
"event_name": "subscriber.updated_lead_score",
"payload":
"event": "subscriber.updated_custom_field",
"data.subscriber.id": "...",
"occurred_at": "2017-10-15T05:36:57.000Z",
"data.account_id": "11111",
"data.subscriber.status": "active",
"data.subscriber.custom_fields.coupon": "xxxxxxx",
"data.subscriber.custom_fields.coupon_discounted_price": "11111",
"data.subscriber.custom_fields.coupon_pre_discount_price": "11111",
"data.subscriber.custom_fields.name": "John Doe",
"data.subscriber.custom_fields.first_name": "John",
"data.subscriber.custom_fields.ip_address": "0.0.0.0",
"data.subscriber.tags": "tag1,tag2,tag3",
"data.subscriber.time_zone": "Europe/Berlin",
"data.subscriber.utc_offset": 120,
"data.subscriber.created_at": "2017-03-27T18:19:35.000Z"
这可能吗? UDF FlattenCustomDimensions
接受一组项目并将它们作为属性公开。 UDF ExtractJsonWebhookPayload
采用字符串并将其转换为 JSON,而 UDF FlattenWebhookPayload
采用复杂的 JSON 对象并创建您在结果中的 payload
对象中看到的点语法。
我的最终目标是获得如下所示的结果集:
"partitionkey": "zzzzzzzzz",
"rowkey": "8beeb783-b07f-8a98-ef56-71c43378a5fc",
"event_time": "2017-10-15T05:37:06.3240000Z",
"event_name": "subscriber.updated_lead_score",
"payload.event": "subscriber.updated_custom_field",
"payload.data.subscriber.id": "...",
"payload.occurred_at": "2017-10-15T05:36:57.000Z",
"payload.data.account_id": "11111",
"payload.data.subscriber.status": "active",
"payload.data.subscriber.custom_fields.coupon": "xxxxxxx",
"payload.data.subscriber.custom_fields.coupon_discounted_price": "11111",
"payload.data.subscriber.custom_fields.coupon_pre_discount_price": "11111",
"payload.data.subscriber.custom_fields.name": "John Doe",
"payload.data.subscriber.custom_fields.first_name": "John",
"payload.data.subscriber.custom_fields.ip_address": "0.0.0.0",
"payload.data.subscriber.tags": "tag1,tag2,tag3",
"payload.data.subscriber.time_zone": "Europe/Berlin",
"payload.data.subscriber.utc_offset": 120,
"payload.data.subscriber.created_at": "2017-03-27T18:19:35.000Z"
除非有人有更好的想法/选择。
【问题讨论】:
如果您知道所有列名,则可以编写查询来提升所有嵌套字段。但是查询将很大并且难以更改。您是否考虑过使用javascript UDF?它会更简洁,您必须在单个字段中传递完整的有效负载,然后将其传递给 UDF。 嗯,是的,重点是我不知道这些领域。知道字段名称很容易通过使用点语法的直接查询来解决。 【参考方案1】:如果将点语法添加到查询中,在最后一行中选择 *,则可以将在 temp 中展开的列查询到主表中的特定列中
【讨论】:
以上是关于在流分析查询中将已知属性从记录提升到***结果的主要内容,如果未能解决你的问题,请参考以下文章