U-sql: Как обработать файл Avro с несколькими массивами JSON с несколькими объектами? - PullRequest
Новая
       45

U-sql: Как обработать файл Avro с несколькими массивами JSON с несколькими объектами?

0 голосов
/ 24 сентября 2018

Я получаю файл Avro в моем хранилище озера данных через потоковую аналитику и концентратор событий с использованием захвата.

Структура файла выглядит следующим образом:

[{"id":1, "PID": "ABC", "значение": "1", "utctimestamp": 1537805867}, { "идентификатор": 6569, "PID": "1E014000", "значение": "- 5,8",»utctimestamp ": 1537805867}] [{" id ": 2," pid ":" cde "," value ":" 77 "," utctimestamp ": 1537772095}, {" id ": 6658," pid ":" 02002001"," value ":" 77 "," utctimestamp ": 1537772095}]

Образец файла

Я использовал этот скрипт: 

@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);

Сценарий приводит к созданию файла, но только с разделением запятыми в нем и без значений.

Как извлечь / преобразовать эту структуру, чтобы я мог вывести ее в виде плоского csv-файла с 4 столбцами?

1 Ответ

0 голосов
/ 01 октября 2018

Я получил это для работы, снова взорвав столбец 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);

Мои результаты:

Results

...