Megjegyzés
Az oldalhoz való hozzáféréshez engedély szükséges. Megpróbálhat bejelentkezni vagy módosítani a címtárat.
Az oldalhoz való hozzáféréshez engedély szükséges. Megpróbálhatja módosítani a címtárat.
Ebben a cikkben adatot az Azure Data Explorer Python könyvtárával visz be. Az Azure Data Explorer egy gyors és nagy mértékben méretezhető adatfeltárási szolgáltatás a napló- és telemetriaadatokhoz. Az Azure Data Explorer két ügyfélkódtárat biztosít a Pythonhoz: egy betöltési tárat és egy adattárat. Ezek a kódtárak lehetővé teszik adatok betöltését vagy betöltését egy fürtbe, és adatokat kérdezhet le a kódból.
Először hozzon létre egy táblát és egy adatleképezést egy fürtben. Ezután várólista-betöltést kell végrehajtania a fürtnek, és ellenőriznie kell az eredményeket.
Előfeltételek
- Microsoft-fiók vagy Microsoft Entra felhasználói identitás. Nincs szükség Azure-előfizetésre.
- Egy Azure Data Explorer-fürt és adatbázis. Hozzon létre egy klasztert és egy adatbázist.
- Python 3.4+.
Az adatok és a beviteli könyvtárak telepítése
Telepítse az azure-kusto-data és az azure-kusto-ingest csomagokat.
pip install azure-kusto-data
pip install azure-kusto-ingest
Importálási utasítások és állandók hozzáadása
Osztályok importálása az azure-kusto-data csomagból.
from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
from azure.kusto.data.exceptions import KustoServiceError
from azure.kusto.data.helpers import dataframe_from_result_table
Egy alkalmazás hitelesítéséhez az Azure Data Explorer a Microsoft Entra-bérlő azonosítóját használja. A bérlőazonosítójának megtalálásához használja az alábbi URL-t, és cserélje le a domain nevét az ÖnDomaingel.
https://login.microsoftonline.com/<YourDomain>/.well-known/openid-configuration/
Ha például a domainje contoso.com, az URL-cím a következő: https://login.microsoftonline.com/contoso.com/.well-known/openid-configuration/. Kattintson erre az URL-címre az eredmények megtekintéséhez; az első sor a következő.
"authorization_endpoint":"https://login.microsoftonline.com/6babcaad-604b-40ac-a9d7-9fd97c0b779f/oauth2/authorize"
Ebben az esetben a bérlőazonosító a következő aaaabbbb-0000-cccc-1111-dddd2222eeee: . A kód futtatása előtt állítsa be a AAD_TENANT_ID, KUSTO_URI, KUSTO_INGEST_URI és KUSTO_DATABASE értékeit.
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>"
Most már hozza létre a kapcsolati karakterláncot. Az alábbi példa eszközhitelesítést használ a fürt eléréséhez. Használhat felügyelt identitáshitelesítést, Microsoft Entra-alkalmazástanúsítványt, Microsoft Entra-alkalmazáskulcsot és Microsoft Entra-felhasználót és -jelszót is.
A céltáblát és a leképezést egy későbbi lépésben hozhatja létre.
KCSB_INGEST = KustoConnectionStringBuilder.with_interactive_login(
KUSTO_INGEST_URI)
KCSB_DATA = KustoConnectionStringBuilder.with_interactive_login(
KUSTO_URI)
DESTINATION_TABLE = "StormEvents"
DESTINATION_TABLE_COLUMN_MAPPING = "StormEvents_CSV_Mapping"
Forrásfájl adatainak megadása
Importálja a további osztályokat, és állítsa be az adatforrásfájl állandóit. Ez a példa egy Azure Blob Storage-ban üzemeltetett mintafájlt használ. A StormEvents mintaadatkészlet a National Centers for Environmental Information időjárással kapcsolatos adatait tartalmazza.
from azure.kusto.data import DataFormat
from azure.kusto.ingest import QueuedIngestClient, IngestionProperties, FileDescriptor, BlobDescriptor, DataFormat, ReportLevel, ReportMethod
CONTAINER = "samplefiles"
ACCOUNT_NAME = "kustosamples"
SAS_TOKEN = "" # If relevant add SAS token
FILE_PATH = "StormEvents.csv"
FILE_SIZE = 64158321 # in bytes
BLOB_PATH = "https://" + ACCOUNT_NAME + ".blob.core.windows.net/" + \
CONTAINER + "/" + FILE_PATH + SAS_TOKEN
Hozzon létre egy táblát a fürtön
Hozzon létre egy táblát, amely megfelel a StormEvents.csv fájlban lévő adatok sémájának. A kód futtatásakor a következőhöz hasonló üzenetet ad vissza: A bejelentkezéshez használjon webböngészőt a lap https://microsoft.com/devicelogin megnyitásához, és adja meg a hitelesítéshez F3W4VWZDM kódot. A bejelentkezéshez kövesse a lépéseket, majd térjen vissza a következő kódblokk futtatásához. A kapcsolatot létesítő későbbi kódblokkok ismételt bejelentkezést igényelnek.
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])
Adatbetöltési leképezés meghatározása
Térképezze fel a bejövő CSV-adatokat a tábla létrehozásakor használt oszlopnevekre és adattípusokra. Ez a forrásadatmezőket a céltábla oszlopaihoz rendeli le
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])
Üzenet sorba állítása a feldolgozáshoz
Adatok lekérésére üzenetet sorba állítunk a blobtárolóból, majd betöltjük azokat az Azure Data Explorerbe.
INGESTION_CLIENT = QueuedIngestClient(KCSB_INGEST)
# All ingestion properties are documented here: https://learn.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')
A táblába betöltött adatok lekérdezése
Várjon 5–10 percet, amíg a várólistán lévő betöltés ütemezi a beolvasást, és betölti az adatokat az Azure Data Explorerbe. Ezután futtassa a következő kódot a StormEvents tábla rekordjainak számának lekéréséhez.
QUERY = "StormEvents | count"
RESPONSE = KUSTO_CLIENT.execute_query(KUSTO_DATABASE, QUERY)
dataframe_from_result_table(RESPONSE.primary_results[0])
Hibaelhárítási lekérdezések futtatása
Jelentkezzen be a fürtbe https://dataexplorer.azure.com , és csatlakozzon a fürthöz. Futtassa a következő parancsot az adatbázisban, és ellenőrizze, hogy voltak-e betöltési hibák az elmúlt négy órában. A futtatás előtt cserélje le az adatbázis nevét.
.show ingestion failures
| where FailedOn > ago(4h) and Database == "<DatabaseName>"
Futtassa a következő parancsot az összes betöltési művelet állapotának megtekintéséhez az elmúlt négy órában. A futtatás előtt cserélje le az adatbázis nevét.
.show operations
| where StartedOn > ago(4h) and Database == "<DatabaseName>" and Table == "StormEvents" and Operation == "DataIngestPull"
| summarize arg_max(LastUpdatedOn, *) by OperationId
Erőforrások tisztítása
Ha a többi cikkünket is követni szeretné, tartsa meg a létrehozott erőforrásokat. Ha nem, futtassa a következő parancsot az adatbázisban a StormEvents tábla törléséhez.
.drop table StormEvents