Ingestování dat pomocí Azure Data Explorer Pythonu
V tomto článku budete ingestovat data pomocí Azure Data Explorer Pythonu. Průzkumník dat Azure je rychlá a vysoce škálovatelná služba pro zkoumání dat protokolů a telemetrie. Průzkumník dat Azure nabízí dvě klientské knihovny pro Python: knihovnu ingestů a knihovnu dat. Tyto knihovny umožňují ingestovat nebo načítat data do clusteru a dotazovat se na data z kódu.
Nejprve v clusteru vytvořte tabulku a mapování dat. Pak vytvoříte frontu ingestace do clusteru a ověříte výsledky.
Požadavky
- Předplatné Azure. Vytvořte si bezplatný účet Azure.
- Vytvořte cluster a databázi.
- Python 3.4+.
Instalace knihovny dat a knihovny ingestů
Nainstalujte azure-kusto-data a azure-kusto-ingest.
pip install azure-kusto-data
pip install azure-kusto-ingest
Přidání příkazů a konstant pro import
Import tříd z azure-kusto-data
from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
from azure.kusto.data.exceptions import KustoServiceError
from azure.kusto.data.helpers import dataframe_from_result_table
K ověření aplikace používá Azure Data Explorer ID vašeho tenanta Azure Active Directory. K vyhledání ID tenanta použijte následující adresu URL a nahraďte doménu doménou YourDomain.
https://login.windows.net/<YourDomain>/.well-known/openid-configuration/
Pokud je vaše doména například contoso.com, je adresa URL . Kliknutím na tuto adresu URL zobrazte výsledky. První řádek vypadá jako v následujícím příkladu.
"authorization_endpoint":"https://login.windows.net/6babcaad-604b-40ac-a9d7-9fd97c0b779f/oauth2/authorize"
ID tenanta je v tomto případě 6babcaad-604b-40ac-a9d7-9fd97c0b779f. Před spuštěním tohoto kódu nastavte hodnoty pro AAD_TENANT_ID, KUSTO_URI, KUSTO_INGEST_URI a KUSTO_DATABASE.
AAD_TENANT_ID = "<TenantId>"
KUSTO_URI = "https://<ClusterName>.<Region>.kusto.windows.net/"
KUSTO_INGEST_URI = "https://ingest-<ClusterName>.<Region>.kusto.windows.net/"
KUSTO_DATABASE = "<DatabaseName>"
Teď sestavte připojovací řetězec. V tomto příkladu se pro přístup ke clusteru používá ověřování pomocí zařízení. Můžete také použít Azure Active Directory aplikace,klíč Azure Active Directorya Azure Active Directory uživatelea heslo .
Cílovou tabulku a mapování vytvoříte v pozdějším kroku.
KCSB_INGEST = KustoConnectionStringBuilder.with_aad_device_authentication(
KUSTO_INGEST_URI, AAD_TENANT_ID)
KCSB_DATA = KustoConnectionStringBuilder.with_aad_device_authentication(
KUSTO_URI, AAD_TENANT_ID)
DESTINATION_TABLE = "StormEvents"
DESTINATION_TABLE_COLUMN_MAPPING = "StormEvents_CSV_Mapping"
Nastavení informací o zdrojovém souboru
Importujte další třídy a nastavte konstanty pro soubor zdroje dat. Tento příklad používá ukázkový soubor hostovaný v Azure Blob Storage. Ukázková datová sada StormEvents obsahuje data týkající se počasí od National Centers for Environmental Information.
from azure.kusto.ingest import QueuedIngestClient, IngestionProperties, FileDescriptor, BlobDescriptor, DataFormat, ReportLevel, ReportMethod
CONTAINER = "samplefiles"
ACCOUNT_NAME = "kustosamplefiles"
SAS_TOKEN = "?sv=2019-12-12&ss=b&srt=o&sp=r&se=2022-09-05T02:23:52Z&st=2020-09-04T18:23:52Z&spr=https&sig=VrOfQMT1gUrHltJ8uhjYcCequEcfhjyyMX%2FSc3xsCy4%3D"
FILE_PATH = "StormEvents.csv"
FILE_SIZE = 64158321 # in bytes
BLOB_PATH = "https://" + ACCOUNT_NAME + ".blob.core.windows.net/" + \
CONTAINER + "/" + FILE_PATH + SAS_TOKEN
Vytvoření tabulky v clusteru
Vytvořte tabulku, která odpovídá schématu dat v souboru StormEvents.csv. Při spuštění tohoto kódu se vrátí zpráva podobná následující zprávě: Pokud se chcete přihlásit, otevřete ve webovém prohlížeči stránku a zadejte kód F3W4VWZDMpro ověření . Podle pokynů se přihlaste a pak se vraťte a spusťte další blok kódu. Následující bloky kódu, které provedou připojení, vyžadují, abyste se znovu přihlásili.
KUSTO_CLIENT = KustoClient(KCSB_DATA)
CREATE_TABLE_COMMAND = ".create table StormEvents (StartTime: datetime, EndTime: datetime, EpisodeId: int, EventId: int, State: string, EventType: string, InjuriesDirect: int, InjuriesIndirect: int, DeathsDirect: int, DeathsIndirect: int, DamageProperty: int, DamageCrops: int, Source: string, BeginLocation: string, EndLocation: string, BeginLat: real, BeginLon: real, EndLat: real, EndLon: real, EpisodeNarrative: string, EventNarrative: string, StormSummary: dynamic)"
RESPONSE = KUSTO_CLIENT.execute_mgmt(KUSTO_DATABASE, CREATE_TABLE_COMMAND)
dataframe_from_result_table(RESPONSE.primary_results[0])
Definování mapování ingestace
Namapujte příchozí data CSV na názvy sloupců a datové typy použité při vytváření tabulky. Tím se zdrojová datová pole mapuje na sloupce cílové tabulky.
CREATE_MAPPING_COMMAND = """.create table StormEvents ingestion csv mapping 'StormEvents_CSV_Mapping' '[{"Name":"StartTime","datatype":"datetime","Ordinal":0}, {"Name":"EndTime","datatype":"datetime","Ordinal":1},{"Name":"EpisodeId","datatype":"int","Ordinal":2},{"Name":"EventId","datatype":"int","Ordinal":3},{"Name":"State","datatype":"string","Ordinal":4},{"Name":"EventType","datatype":"string","Ordinal":5},{"Name":"InjuriesDirect","datatype":"int","Ordinal":6},{"Name":"InjuriesIndirect","datatype":"int","Ordinal":7},{"Name":"DeathsDirect","datatype":"int","Ordinal":8},{"Name":"DeathsIndirect","datatype":"int","Ordinal":9},{"Name":"DamageProperty","datatype":"int","Ordinal":10},{"Name":"DamageCrops","datatype":"int","Ordinal":11},{"Name":"Source","datatype":"string","Ordinal":12},{"Name":"BeginLocation","datatype":"string","Ordinal":13},{"Name":"EndLocation","datatype":"string","Ordinal":14},{"Name":"BeginLat","datatype":"real","Ordinal":16},{"Name":"BeginLon","datatype":"real","Ordinal":17},{"Name":"EndLat","datatype":"real","Ordinal":18},{"Name":"EndLon","datatype":"real","Ordinal":19},{"Name":"EpisodeNarrative","datatype":"string","Ordinal":20},{"Name":"EventNarrative","datatype":"string","Ordinal":21},{"Name":"StormSummary","datatype":"dynamic","Ordinal":22}]'"""
RESPONSE = KUSTO_CLIENT.execute_mgmt(KUSTO_DATABASE, CREATE_MAPPING_COMMAND)
dataframe_from_result_table(RESPONSE.primary_results[0])
Přidání zprávy do fronty pro ingestaci
Přidejte zprávu do fronty k získání dat z úložiště objektů blob a tato data ingestujte do Průzkumníka dat Azure.
INGESTION_CLIENT = QueuedIngestClient(KCSB_INGEST)
# All ingestion properties are documented here: https://docs.microsoft.com/azure/kusto/management/data-ingest#ingestion-properties
INGESTION_PROPERTIES = IngestionProperties(database=KUSTO_DATABASE, table=DESTINATION_TABLE, data_format=DataFormat.CSV,
ingestion_mapping_reference=DESTINATION_TABLE_COLUMN_MAPPING, additional_properties={'ignoreFirstRecord': 'true'})
# FILE_SIZE is the raw size of the data in bytes
BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
INGESTION_CLIENT.ingest_from_blob(
BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)
print('Done queuing up ingestion with Azure Data Explorer')
Dotazování na data ingestovaná do tabulky
Počkejte pět až 10 minut, než ingestování ve frontě naplánuje příjem dat a načte data do Azure Data Explorer. Pak spuštěním následujícího kódu získejte počet záznamů v tabulce StormEvents.
QUERY = "StormEvents | count"
RESPONSE = KUSTO_CLIENT.execute_query(KUSTO_DATABASE, QUERY)
dataframe_from_result_table(RESPONSE.primary_results[0])
Spuštění dotazů pro řešení potíží
Přihlaste se k https://dataexplorer.azure.com a připojte se k vašemu clusteru. Spuštěním následujícího příkazu ve vaší databázi zjistíte, jestli za poslední čtyři hodiny došlo k chybám ingestování. Přes spuštěním nahraďte název databáze.
.show ingestion failures
| where FailedOn > ago(4h) and Database == "<DatabaseName>"
Spuštěním následujícího příkazu zobrazíte stav všech operací ingestace za poslední čtyři hodiny. Přes spuštěním nahraďte název databáze.
.show operations
| where StartedOn > ago(4h) and Database == "<DatabaseName>" and Table == "StormEvents" and Operation == "DataIngestPull"
| summarize arg_max(LastUpdatedOn, *) by OperationId
Vyčištění prostředků
Pokud máte v plánu postupovat podle našich dalších článků, pořiďte si prostředky, které jste vytvořili. Pokud ne, spuštěním následujícího příkazu v databázi tabulku StormEvents vyčistěte.
.drop table StormEvents