相关文章推荐
小百科  ›  將 JSON 格式化的資料內嵌至 Azure Data Explorer - Azure Data Explorer | Microsoft Learn
table properties string
暴躁的四季豆
2 年前

本文說明如何將 JSON 格式化的資料內嵌至 Azure Data Explorer 資料庫。 您將從原始和對應的 JSON 的簡單範例開始，繼續多行 JSON，然後處理包含陣列和字典的更複雜的 JSON 架構。 這些範例詳細說明使用 Kusto 查詢語言 (KQL) 、C# 或 Python 擷取 JSON 格式化資料的程式。 Kusto 查詢語言 ingest 管理命令會直接執行至引擎端點。 在生產案例中，擷取會使用用戶端程式庫或資料連線來執行至資料管理服務。 使用 Azure Data Explorer Python 程式庫 讀取擷取資料，並使用 Azure Data Explorer .NET Standard SDK 擷取資料，以取得使用這些用戶端程式庫內嵌資料的逐步解說。

  • Microsoft 帳戶或 Azure Active Directory 使用者身分識別。 不需要 Azure 訂用帳戶。
  • Azure 資料總管叢集和資料庫。 建立叢集和資料庫 。

    • JSON 格式

    Azure Data Explorer支援兩種 JSON 檔案格式：

  • json ：以行分隔的 JSON。 輸入資料中的每個行都只有一個 JSON 記錄。 此格式支援剖析批註和單引號屬性。 如需詳細資訊，請參閱 JSON 行 。
  • multijson ：多行 JSON。 剖析器會忽略行分隔符號，並將前一個位置的記錄讀取到有效 JSON 的結尾。 此格式支援剖析批註、單引號屬性和分行符號。

    • 使用擷 取精靈擷取 時，預設格式為 multijson 。 格式可以處理多行 JSON 記錄和 JSON 記錄的陣列。 遇到剖析錯誤時，會捨棄整個檔案。

    如果您使用 JSON Line 格式，其中每一行都是單一格式正確的 JSON 記錄，而且您想要能夠處理格式不正確之記錄，您可以選取 [忽略資料格式錯誤] 選項。這會允許在略過格式不正確時擷取有效的記錄。

    擷取和對應 JSON 格式化的資料

    擷取 JSON 格式化資料需要您使用 擷取屬性 來指定 格式 。 擷取 JSON 資料需要 對應 ，這會將 JSON 來源專案對應至其目標資料行。 擷取資料時，請使用 IngestionMapping 屬性搭配其 ingestionMappingReference (預先定義的對應) 擷取屬性或其 IngestionMappings 屬性。 本文將使用 ingestionMappingReference 擷取屬性，該屬性是在用於擷取的資料表上預先定義。 在下列範例中，我們將先將 JSON 記錄擷取為原始資料到單一資料行資料表。 然後，我們將使用 對應，將每個屬性內嵌至其對應的資料行。

    簡單 JSON 範例

    下列範例是具有一般結構的簡單 JSON。 資料具有數個裝置所收集的溫度和濕度資訊。 每個記錄都會以識別碼和時間戳記標示。

    "timestamp": "2019-05-02 15:23:50.0369439", "deviceId": "2945c8aa-f13e-4c48-4473-b81440bb5ca2", "messageId": "7f316225-839a-4593-92b5-1812949279b3", "temperature": 31.0301639051317, "humidity": 62.0791099602725

    擷取原始 JSON 記錄

    在此範例中，您會將 JSON 記錄擷取為原始資料至單一資料行資料表。 資料擷取之後，就會執行資料操作、使用查詢和更新原則。

    Python

  • 選取 [新增叢集]。

  • 在 [新增叢集] 對話方塊中，以 https://<ClusterName>.<Region>.kusto.windows.net/ 格式輸入您的叢集 URL，然後選取 [新增] 。

  • 貼入下列命令，並選取 [執行] 以建立資料表。 

    .create table RawEvents (Event: dynamic)

    此查詢會建立具有動態資料類型之單 Event 一資料行的資料表。
    

  • 建立 JSON 對應。
    

    .create table RawEvents ingestion json mapping 'RawEventMapping' '[{"column":"Event","Properties":{"path":"$"}}]'

    此命令會建立對應，並將 JSON 根路徑 $ 對應至資料 Event 行。
    

  • 將資料內嵌到 RawEvents 資料表中。
    

    .ingest into table RawEvents ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') with '{"format":"json", "ingestionMappingReference":"RawEventMapping"}'

  • RawEvents建立資料表。
    

    var kustoUri = "https://<clusterName>.<region>.kusto.windows.net/";
var connectionStringBuilder = new KustoConnectionStringBuilder(kustoUri)
    FederatedSecurity = true,
    UserID = userId,
    Password = password,
    Authority = tenantId,
    InitialCatalog = databaseName
using var kustoClient = KustoClientFactory.CreateCslAdminProvider(connectionStringBuilder);
var tableName = "RawEvents";
var command = CslCommandGenerator.GenerateTableCreateCommand(
    tableName,
    new[] { Tuple.Create("Events", "System.Object") }
await kustoClient.ExecuteControlCommandAsync(command);

  • 建立 JSON 對應。
    

    var tableMappingName = "RawEventMapping";
command = CslCommandGenerator.GenerateTableMappingCreateCommand(
    IngestionMappingKind.Json,
    tableName,
    tableMappingName,
    new ColumnMapping[]
        new() { ColumnName = "Events", Properties = new Dictionary<string, string> { { "path", "$" } } }
await kustoClient.ExecuteControlCommandAsync(command);

    此命令會建立對應，並將 JSON 根路徑 $ 對應至資料 Event 行。
    

  • 將資料內嵌到 RawEvents 資料表中。
    

    var ingestUri = "https://ingest-<clusterName>.<region>.kusto.windows.net/";
var ingestConnectionStringBuilder = new KustoConnectionStringBuilder(ingestUri)
    FederatedSecurity = true,
    UserID = userId,
    Password = password,
    Authority = tenantId,
    InitialCatalog = databaseName
using var ingestClient = KustoIngestFactory.CreateQueuedIngestClient(ingestConnectionStringBuilder);
var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json";
var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
    Format = DataSourceFormat.json,
    IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
await ingestClient.IngestFromStorageAsync(blobPath, properties);

  • RawEvents建立資料表。
    

    KUSTO_URI = "https://<ClusterName>.<Region>.kusto.windows.net/"
KCSB_DATA = KustoConnectionStringBuilder.with_aad_device_authentication(KUSTO_URI, AAD_TENANT_ID)
KUSTO_CLIENT = KustoClient(KCSB_DATA)
TABLE = "RawEvents"
CREATE_TABLE_COMMAND = ".create table " + TABLE + " (Events: dynamic)"
RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_TABLE_COMMAND)
dataframe_from_result_table(RESPONSE.primary_results[0])

  • 建立 JSON 對應。
    

    MAPPING = "RawEventMapping"
CREATE_MAPPING_COMMAND = ".create table " + TABLE + " ingestion json mapping '" + MAPPING + """' '[{"column":"Event","path":"$"}]'"""
RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_MAPPING_COMMAND)
dataframe_from_result_table(RESPONSE.primary_results[0])

  • 將資料內嵌到 RawEvents 資料表中。
    

    INGEST_URI = "https://ingest-<ClusterName>.<Region>.kusto.windows.net/"
KCSB_INGEST = KustoConnectionStringBuilder.with_aad_device_authentication(INGEST_URI, AAD_TENANT_ID)
INGESTION_CLIENT = KustoIngestClient(KCSB_INGEST)
BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json'
INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.JSON, ingestion_mapping_reference=MAPPING)
BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
INGESTION_CLIENT.ingest_from_blob(
    BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)

    資料會根據 批次處理原則進行匯總，導致幾分鐘的延遲。
    

  • 建立新的資料表，其架構與 JSON 輸入資料類似。 我們將針對下列所有範例和擷取命令使用此資料表。
    

    .create table Events (Time: datetime, Device: string, MessageId: string, Temperature: double, Humidity: double)

  • 建立 JSON 對應。
    

    .create table Events ingestion json mapping 'FlatEventMapping' '[{"column":"Time","Properties":{"path":"$.timestamp"}},{"column":"Device","Properties":{"path":"$.deviceId"}},{"column":"MessageId","Properties":{"path":"$.messageId"}},{"column":"Temperature","Properties":{"path":"$.temperature"}},{"column":"Humidity","Properties":{"path":"$.humidity"}}]'

    在此對應中，如資料表架構所定義，專案 timestamp 會內嵌至資料行 Time 作為 datetime 資料類型。
    

  • 將資料內嵌到 Events 資料表中。
    

    .ingest into table Events ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') with '{"format":"json", "ingestionMappingReference":"FlatEventMapping"}'

    檔案 'simple.json' 有一些以行分隔的 JSON 記錄。 格式為 json ，而擷取命令中使用的對應就是 FlatEventMapping 您所建立的。
    

  • 建立新的資料表，其架構與 JSON 輸入資料類似。 我們將針對下列所有範例和擷取命令使用此資料表。
    

    var tableName = "Events";
var command = CslCommandGenerator.GenerateTableCreateCommand(
   tableName,
   new[]
       Tuple.Create("Time", "System.DateTime"),
       Tuple.Create("Device", "System.String"),
       Tuple.Create("MessageId", "System.String"),
       Tuple.Create("Temperature", "System.Double"),
       Tuple.Create("Humidity", "System.Double")
await kustoClient.ExecuteControlCommandAsync(command);

  • 建立 JSON 對應。
    

    var tableMappingName = "FlatEventMapping";
command = CslCommandGenerator.GenerateTableMappingCreateCommand(
    IngestionMappingKind.Json,
    tableName,
    tableMappingName,
    new ColumnMapping[]
        new() { ColumnName = "Time", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.timestamp" } } },
        new() { ColumnName = "Device", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.deviceId" } } },
        new() { ColumnName = "MessageId", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.messageId" } } },
        new() { ColumnName = "Temperature", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.temperature" } } },
        new() { ColumnName = "Humidity", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.humidity" } } }
await kustoClient.ExecuteControlCommandAsync(command);

    在此對應中，如資料表架構所定義，專案 timestamp 會內嵌至資料行 Time 作為 datetime 資料類型。
    

  • 將資料內嵌到 Events 資料表中。
    

    var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json";
var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
    Format = DataSourceFormat.json,
    IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
await ingestClient.IngestFromStorageAsync(blobPath, properties).ConfigureAwait(false);

    檔案 'simple.json' 有一些以行分隔的 JSON 記錄。 格式為 json ，而擷取命令中使用的對應就是 FlatEventMapping 您所建立的。
    

  • 建立新的資料表，其架構與 JSON 輸入資料類似。 我們將針對下列所有範例和擷取命令使用此資料表。
    

    TABLE = "Events"
CREATE_TABLE_COMMAND = ".create table " + TABLE + " (Time: datetime, Device: string, MessageId: string, Temperature: double, Humidity: double)"
RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_TABLE_COMMAND)
dataframe_from_result_table(RESPONSE.primary_results[0])

  • 建立 JSON 對應。
    

    MAPPING = "FlatEventMapping"
CREATE_MAPPING_COMMAND = ".create table Events ingestion json mapping '" + MAPPING + """' '[{"column":"Time","Properties":{"path":"$.timestamp"}},{"column":"Device","Properties":{"path":"$.deviceId"}},{"column":"MessageId","Properties":{"path":"$.messageId"}},{"column":"Temperature","Properties":{"path":"$.temperature"}},{"column":"Humidity","Properties":{"path":"$.humidity"}}]'"""
RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_MAPPING_COMMAND)
dataframe_from_result_table(RESPONSE.primary_results[0])

  • 將資料內嵌到 Events 資料表中。
    

    BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json'
INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.JSON, ingestion_mapping_reference=MAPPING)
BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
INGESTION_CLIENT.ingest_from_blob(
    BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)

    檔案 'simple.json' 有幾行分隔的 JSON 記錄。 格式為 json ，而擷取命令中使用的對應就是 FlatEventMapping 您所建立的。
    

    將資料內嵌到 Events 資料表中。
    

    var tableMappingName = "FlatEventMapping";
var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/multilined.json";
var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
    Format = DataSourceFormat.multijson,
    IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
await ingestClient.IngestFromStorageAsync(blobPath, properties).ConfigureAwait(false);

    將資料內嵌到 Events 資料表中。
    

    MAPPING = "FlatEventMapping"
BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/multilined.json'
INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.MULTIJSON, ingestion_mapping_reference=MAPPING)
BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
INGESTION_CLIENT.ingest_from_blob(
    BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)

    內嵌包含陣列的 JSON 記錄
    

    陣列資料類型是已排序的值集合。 JSON 陣列的擷取是由 更新原則所完成。 JSON 會依原狀擷取至中繼資料表。 更新原則會在資料表上 RawEvents 執行預先定義的函式，並將結果重新內嵌至目標資料表。 我們將內嵌具有下列結構的資料：
    
    "records":
            "timestamp": "2019-05-02 15:23:50.0000000",
            "deviceId": "ddbc1bf5-096f-42c0-a771-bc3dca77ac71",
            "messageId": "7f316225-839a-4593-92b5-1812949279b3",
            "temperature": 31.0301639051317,
            "humidity": 62.0791099602725
            "timestamp": "2019-05-02 15:23:51.0000000",
            "deviceId": "ddbc1bf5-096f-42c0-a771-bc3dca77ac71",
            "messageId": "57de2821-7581-40e4-861e-ea3bde102364",
            "temperature": 33.7529423105311,
            "humidity": 75.4787976739364

  • 建立可 update policy 展開 集合 records 的函式，讓集合中的每個值都使用 mv-expand 運算子接收個別的資料列。 我們將使用資料表 RawEvents 作為來源資料表和 Events 目標資料表。
    

    .create function EventRecordsExpand() {
    RawEvents
    | mv-expand records = Event.records
    | project
        Time = todatetime(records["timestamp"]),
        Device = tostring(records["deviceId"]),
        MessageId = tostring(records["messageId"]),
        Temperature = todouble(records["temperature"]),
        Humidity = todouble(records["humidity"])

  • 函式收到的架構必須符合目標資料表的架構。 使用 getschema 運算子來檢閱架構。
    

    EventRecordsExpand() | getschema

  • 將更新原則新增至目標資料表。 此原則會自動對中繼資料表中 RawEvents 任何新內嵌的資料執行查詢，並將結果內嵌到資料表中 Events 。 定義零保留原則，以避免保存中繼資料表。
    

    .alter table Events policy update @'[{"Source": "RawEvents", "Query": "EventRecordsExpand()", "IsEnabled": "True"}]'

  • 將資料內嵌到 RawEvents 資料表中。
    

    .ingest into table RawEvents ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json') with '{"format":"multijson", "ingestionMappingReference":"RawEventMapping"}'

  • 檢閱資料表中的資料 Events 。
    

    Events

  • 建立可展開 集合 records 的更新函式，讓集合中的每個值都使用 mv-expand 運算子接收個別的資料列。 我們將使用資料表 RawEvents 作為來源資料表和 Events 目標資料表。
    

    var command = CslCommandGenerator.GenerateCreateFunctionCommand(
    "EventRecordsExpand",
    "UpdateFunctions",
    string.Empty,
    null,
    @"RawEvents
    | mv-expand records = Event
    | project
        Time = todatetime(records['timestamp']),
        Device = tostring(records['deviceId']),
        MessageId = tostring(records['messageId']),
        Temperature = todouble(records['temperature']),
        Humidity = todouble(records['humidity'])",
    ifNotExists: false
await kustoClient.ExecuteControlCommandAsync(command);

    函式收到的架構必須符合目標資料表的架構。
    

  • 將更新原則新增至目標資料表。 此原則會自動對中繼資料表中 RawEvents 任何新內嵌的資料執行查詢，並將其結果內嵌到資料表中 Events 。 定義零保留原則，以避免保存中繼資料表。
    

    command = ".alter table Events policy update @'[{'Source': 'RawEvents', 'Query': 'EventRecordsExpand()', 'IsEnabled': 'True'}]";
await kustoClient.ExecuteControlCommandAsync(command);

  • 將資料內嵌到 RawEvents 資料表中。
    

    var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json";
var tableName = "RawEvents";
var tableMappingName = "RawEventMapping";
var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
    Format = DataSourceFormat.multijson,
    IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
await ingestClient.IngestFromStorageAsync(blobPath, properties);

  • 檢閱資料表中的資料 Events 。
    

  • 建立可展開 集合 records 的更新函式，讓集合中的每個值都使用 mv-expand 運算子接收個別的資料列。 我們將使用資料表 RawEvents 作為來源資料表和 Events 目標資料表。
    

    CREATE_FUNCTION_COMMAND =
    '''.create function EventRecordsExpand() {
        RawEvents
        | mv-expand records = Event
        | project
            Time = todatetime(records["timestamp"]),
            Device = tostring(records["deviceId"]),
            MessageId = tostring(records["messageId"]),
            Temperature = todouble(records["temperature"]),
            Humidity = todouble(records["humidity"])
RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_FUNCTION_COMMAND)
dataframe_from_result_table(RESPONSE.primary_results[0])

    函式收到的架構必須符合目標資料表的架構。
    

  • 將更新原則新增至目標資料表。 此原則會自動對中繼資料表中 RawEvents 任何新內嵌的資料執行查詢，並將其結果內嵌到資料表中 Events 。 定義零保留原則，以避免保存中繼資料表。
    

    CREATE_UPDATE_POLICY_COMMAND =
    """.alter table Events policy update @'[{'Source': 'RawEvents', 'Query': 'EventRecordsExpand()', 'IsEnabled': 'True'}]"""
RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_UPDATE_POLICY_COMMAND)
dataframe_from_result_table(RESPONSE.primary_results[0])

  • 將資料內嵌到 RawEvents 資料表中。
    

    TABLE = "RawEvents"
MAPPING = "RawEventMapping"
BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json'
INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.MULTIJSON, ingestion_mapping_reference=MAPPING)
BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
INGESTION_CLIENT.ingest_from_blob(
    BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)

  • 檢閱資料表中的資料 Events 。
    •  
    推荐文章
    今天看啥   ·   Py中国   ·   codingpro   ·   小百科   ·   link之家   ·   卧龙AI搜索
    删除内容请联系邮箱 2879853325@qq.com
    小百科 - 百科知识指南
    © 2024 ~ 沪ICP备11025650号