U-sql：如何处理具有多个 JSON 数组和多个对象的 Avro 文件？答案

【问题标题】：U-sql : How to process an Avro file with multiple JSON arrays with multiple objects?U-sql：如何处理具有多个 JSON 数组和多个对象的 Avro 文件？
【发布时间】：2018-09-24 12:51:04
【问题描述】：

我通过流式分析和使用捕获的事件中心在我的 Data Lake Store 中收到 Avro 文件。

文件结构如下：

[{"id":1,"pid":"abc","value":"1","utctimestamp":1537805867},{"id":6569,"pid":"1E014000","值":"-5.8","utctimestamp":1537805867}] [{"id":2,"pid":"cde","value":"77","utctimestamp":1537772095},{"id":6658,"pid":"02002001","value": "77","utctimestamp":1537772095}]

Sample File

我用过这个脚本：

@rs =
EXTRACT
    SequenceNumber      long,
    Offset              string,
    EnqueuedTimeUtc     string,
    Body                byte[]
FROM @input_file
USING new Microsoft.Analytics.Samples.Formats.ApacheAvro.AvroExtractor(@"
{
""type"": ""record"",
""name"": ""EventData"",
""namespace"": ""Microsoft.ServiceBus.Messaging"",
""fields"": [
    {
        ""name"": ""SequenceNumber"",
        ""type"": ""long""
    },
    {
        ""name"": ""Offset"",
        ""type"": ""string""
    },
    {
        ""name"": ""EnqueuedTimeUtc"",
        ""type"": ""string""
    },
    {
        ""name"": ""SystemProperties"",
        ""type"": {
            ""type"": ""map"",
            ""values"": [
                ""long"",
                ""double"",
                ""string"",
                ""bytes""
            ]
        }
    },
    {
        ""name"": ""Properties"",
        ""type"": {
            ""type"": ""map"",
            ""values"": [
                ""long"",
                ""double"",
                ""string"",
                ""bytes"",
                ""null""
            ]
        }
    },
    {
        ""name"": ""Body"",
        ""type"": [
            ""null"",
            ""bytes""
        ]
    }
]
}
");

@jsonify = SELECT Microsoft.Analytics.Samples.Formats.Json.JsonFunctions.JsonTuple(Encoding.UTF8.GetString(Body)) AS message FROM @rs;

@cnt =  SELECT  message["id"] AS id,
            message["id2"] AS pid,
            message["value"] AS value,
            message["utctimestamp"] AS utctimestamp,
            message["extra"] AS extra
    FROM @jsonify;

OUTPUT @cnt TO @output_file USING Outputters.Text(quoting: false);

脚本生成一个文件，但其中只有逗号分隔，没有值。

如何提取/转换此结构，以便将其输出为扁平的 4 列 csv 文件？

【问题讨论】：

如果没有示例文件，这很难重现 - 例如，您可以将两个记录示例从上面保存到 gist.github.com 吗？
@wBob 我添加了一个示例文件。希望你能帮我解决这个问题。

标签： u-sql azure-eventhub-capture streaming-analytics

【解决方案1】：

我通过再次分解 JSON 列并再次应用 JsonTuple 函数来实现这一点（但我怀疑它可以被简化）：

@jsonify =
    SELECT JsonFunctions.JsonTuple(Encoding.UTF8.GetString(Body)) AS message
    FROM @rs;

// Explode the tuple as key-value pair;
@working =
    SELECT key,
           JsonFunctions.JsonTuple(value) AS value
    FROM @jsonify
         CROSS APPLY
             EXPLODE(message) AS y(key, value);

完整脚本：

REFERENCE ASSEMBLY Avro;
REFERENCE ASSEMBLY [Newtonsoft.Json];
REFERENCE ASSEMBLY [Microsoft.Analytics.Samples.Formats]; 

USING Microsoft.Analytics.Samples.Formats.Json;

DECLARE @input_file string = @"\input\input21.avro";
DECLARE @output_file string = @"\output\output.csv";


@rs =
EXTRACT
 Body byte[]
FROM @input_file
USING new Microsoft.Analytics.Samples.Formats.ApacheAvro.AvroExtractor(@"{
    ""type"": ""record"",
    ""name"": ""EventData"",
    ""namespace"": ""Microsoft.ServiceBus.Messaging"",
    ""fields"": [
        {
            ""name"": ""SequenceNumber"",
            ""type"": ""long""
        },
        {
            ""name"": ""Offset"",
            ""type"": ""string""
        },
        {
            ""name"": ""EnqueuedTimeUtc"",
            ""type"": ""string""
        },
        {
            ""name"": ""SystemProperties"",
            ""type"": {
                ""type"": ""map"",
                ""values"": [
                    ""long"",
                    ""double"",
                    ""string"",
                    ""bytes""
                ]
            }
        },
        {
            ""name"": ""Properties"",
            ""type"": {
                ""type"": ""map"",
                ""values"": [
                    ""long"",
                    ""double"",
                    ""string"",
                    ""bytes"",
                    ""null""
                ]
            }
        },
        {
            ""name"": ""Body"",
            ""type"": [
                ""null"",
                ""bytes""
            ]
        }
    ]
}");


@jsonify =
    SELECT JsonFunctions.JsonTuple(Encoding.UTF8.GetString(Body)) AS message
    FROM @rs;

// Explode the tuple as key-value pair;
@working =
    SELECT key,
           JsonFunctions.JsonTuple(value) AS value
    FROM @jsonify
         CROSS APPLY
             EXPLODE(message) AS y(key, value);


@cnt =
    SELECT value["id"] AS id,
           value["id2"] AS pid,
           value["value"] AS value,
           value["utctimestamp"] AS utctimestamp,
           value["extra"] AS extra
    FROM @working;


OUTPUT @cnt TO @output_file USING Outputters.Text(quoting: false);

我的结果：

【讨论】：