我需要将下面的 C# 代码翻译成 Azure 流分析可以使用的东西。
我有一个类似于以下内容的 C# 应用程序:
var inputEvents = new List<Event>();
foreach (var file in files){
(List<Event> events, DateTime maxDate) = ProcessEvents(file, inputEvents);
inputEvents = events.Where(e => e.Duration == null).ToList();
}
ProcessEvents() 将 inputEvents 传递给其他辅助方法的位置
我需要使用流分析来实现整个代码。var 文件部分是通过使用 Collect() 收集一堆事件来实现的。每个批次都发送到充当ProcessEvents()的 UDF 。但是,ProcessEvents() 返回下一次迭代所需的其他事件。由于 UDF 是无状态的,因此下一批将无法使用上一批返回的事件。
如何在流分析中重写上面的 C# 代码?
我尝试了以下方法:
- 使用 UDA 存储返回的事件。失败,因为由于某种原因它无法存储 JSON 数组并不断修改它。
- 使用参考数据输入来存储返回的事件。失败,因为它们只能在使用 JOIN 的流分析中使用,而不能在 UDF 中使用。
流分析 T-SQL 查询:
WITH eventsCollection AS (SELECT COLLECT() AS allEvents
FROM EventHubStreamMessage
GROUP BY SessionWindow(minute,2,4)),
step1 AS (
SELECT UDF.SampleProcessEvents(allEvents) as Source
FROM eventsCollection
)
SELECT *
INTO [StorageTable]
FROM step1
流分析 UDF 代码(短版):
function main(allEvents){
allEvents = JSON.stringify(allEvents);
var inputEvents = new Array ();
return processEvents(JSON.parse(allEvents), inputEvents)
}
function processEvents(allEvents, inputEvents){
for (i=0l i<allEvents.length; i++){
if (allEvents[i].Event =="ON"){
powerOn(inputEvents);
}
}
}
function powerOn(inputEvents){
return true;
}