Erfassen von Daten aus Cribl Stream in Azure Data Explorer

  • Artikel

Cribl Stream ist eine Verarbeitungs-Engine, die Computerereignisdaten aus beliebigen Quellen auf sichere Weise sammelt, verarbeitet und streamt. Sie können diese Daten für jedes Ziel zur Analyse und Verwaltung auf sichere Weise parsen und verarbeiten.

In diesem Artikel wird gezeigt, wie Sie Daten mit Cribl Stream erfassen.

Eine vollständige Liste der Datenconnectors finden Sie in der Übersicht über Datenintegrationen.

Voraussetzungen

Erstellen eines Microsoft Entra-Dienstprinzipals

Der Microsoft Entra-Dienstprinzipal kann über das Azure-Portal oder programmgesteuert (wie im folgenden Beispiel) erstellt werden.

Dieser Dienstprinzipal ist die Identität, die vom Connector zum Schreiben von Daten in Ihre Tabelle in Kusto verwendet wird. Sie gewähren diesem Dienstprinzipal Berechtigungen für den Zugriff auf Kusto-Ressourcen.

  1. Melden Sie sich per Azure CLI an Ihrem Azure-Abonnement an. Führen Sie anschließend im Browser die Authentifizierung durch.

    az login

  2. Wählen Sie das Abonnement aus, um den Prinzipal zu hosten. Dieser Schritt ist erforderlich, wenn Sie über mehrere Abonnements verfügen.

    az account set --subscription YOUR_SUBSCRIPTION_GUID

  3. Erstellen Sie den Dienstprinzipal. In diesem Beispiel wird der Dienstprinzipal als my-service-principal bezeichnet.

    az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}

  4. Kopieren Sie aus den zurückgegebenen JSON-Daten appId, password und tenant für die zukünftige Verwendung.

    {
  "appId": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "displayName": "my-service-principal",
  "name": "my-service-principal",
  "password": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "tenant": "00001111-aaaa-2222-bbbb-3333cccc4444"
}

Sie haben Ihre Microsoft Entra-Anwendung und den Dienstprinzipal erstellt.

Erstellen einer Zieltabelle

Erstellen Sie eine Zieltabelle für die eingehenden Daten und eine Erfassungszuordnung, um die erfassten Datenspalten den Spalten in der Zieltabelle zuzuordnen.

  1. Führen Sie den folgenden Tabellenerstellungsbefehl im Abfrage-Editor aus, und ersetzen Sie den Platzhalter TableName durch den Namen der Zieltabelle:

    .create table <TableName> (_raw: string, _time: long, cribl_pipe: dynamic)

  2. Führen Sie den folgenden create ingestion mapping-Befehl aus, und ersetzen Sie dabei den Platzhalter TableName durch den Zieltabellennamen und den Platzhalter TableNameMapping durch den Namen der Erfassungszuordnung:

    .create table <TableName> ingestion csv mapping '<TableNameMapping>' 'CriblLogMapping' '[{"Name":"_raw","DataType":"string","Ordinal":"0","ConstValue":null},{"Name":"_time","DataType":"long","Ordinal":"1","ConstValue":null},{"Name":"cribl_pipe","DataType":"dynamic","Ordinal":"2","ConstValue":null}]'

  3. Gewähren Sie dem Dienstprinzipal aus Microsoft Entra Dienstprinzipal erstellen Datenbankerfasser Rollenberechtigungen, um mit der Datenbank zu arbeiten. Weitere Informationen finden Sie unter Beispiele. Ersetzen Sie DatabaseName durch den Namen der Zieldatenbank und ApplicationID durch den AppId Wert, den Sie beim Erstellen eines Microsoft Entra-Dienstprinzipals gespeichert haben.

    .add database <DatabaseName> ingestors ('aadapp=<ApplicationID>') 'App Registration'

Erstellen von Cribl Stream-Zielen

Im folgenden Abschnitt wird beschrieben, wie Sie ein Cribl Stream-Ziel erstellen, das Daten in Ihre Tabelle in Kusto schreibt. Für jede Tabelle ist ein separater Cribl Stream-Zielconnector erforderlich.

Auswählen des Ziels

So verbinden Sie Cribl Stream mit Ihrer Tabelle

  1. Wählen Sie im oberen Navigationsbereich in Cribl Manage und dann eine Workergruppe aus.

  2. Wählen Sie Routing>QuickConnect (Stream)>Add Destination aus.

  3. Wählen Sie im Fenster Set up new QuickConnect Destination die Option Azure Data Explorer und dann Add now aus.

Festlegen allgemeiner Einstellungen

Legen Sie im Fenster New Data Explorer unter General Settings die folgenden Einstellungen fest:

Einstellung Wert Beschreibung
Output ID <OutputID>, z. B. KustoDestination Der Name, der zum Identifizieren Ihres Ziels verwendet wird
Ingestion Mode Batching (Standard) oder Streaming Die Einstellungen für den Erfassungsmodus. Mit der Batchverarbeitung kann Ihre Tabelle Datenbatches aus einem Cribl-Speichercontainer pullen, wenn große Datenmengen über einen kurzen Zeitraum erfasst werden. Durch Streaming werden Daten direkt an die KQL-Zieltabelle gesendet. Streaming ist nützlich zum Erfassen kleinerer Datenmengen oder z. B. zum Senden einer kritischen Warnung in Echtzeit. Durch Streaming kann eine geringere Wartezeit als durch Batchverarbeitung erreicht werden. Ist der Erfassungsmodus auf Streaming festgelegt, müssen Sie eine Streamingrichtlinie aktivieren. Weitere Informationen finden Sie unter Richtlinie für die Streamingerfassung.
Cluster base URI Basis-URI Der Basis-URI
Ingestion service URI Erfassungs-URI Wird angezeigt, wenn der Modus Batching ausgewählt ist. Der Erfassungs-URI
Datenbankname <DatabaseName> Der Name Ihrer Zieldatenbank.
Tabellenname <TableName> Der Name Ihrer Zieltabelle.
Validate database settings Yes (Standard) oder No Überprüft die Anmeldeinformationen der Dienstprinzipal-App, die Sie beim Speichern oder Starten Ihres Ziels eingegeben haben. Der Tabellenname wird überprüft, außer wenn Add mapping object aktiviert ist. Diese Einstellung sollte deaktiviert werden, wenn Ihre App nicht über die beiden Rollen als anzeigender Datenbankbenutzer und anzeigender Tabellenbenutzer verfügt.
Add mapping object Yes oder No (Standard) Wird nur angezeigt, wenn der Batchverarbeitungsmodus anstelle des standardmäßigen Textfelds Data mapping ausgewählt ist. Wenn Sie Yes auswählen, wird ein Fenster zur Eingabe der Datenzuordnung als JSON-Objekt geöffnet.
Datenzuordnung Der Name des Zuordnungsschemas wie im Schritt Erstellen einer Zieltabelle beschrieben. Der Name des Zuordnungsschemas. Die Standardansicht, wenn Add mapping object auf Nofestgelegt ist.
Komprimieren gzip (Standard) Wenn Data format auf „Parquet“ festgelegt ist, ist die Option Compress nicht verfügbar.
Datenformat „JSON“ (Standard), „Raw“ oder „Parquet“ Das Datenformat. „Parquet“ ist nur im Modus Batching verfügbar und wird nur unter Linux unterstützt.
Backpressure behavior Block (Standard) oder Drop Wählen Sie aus, ob Ereignisse blockiert oder weggelassen werden sollen, wenn Empfänger einen Rückstau aufweisen.
Tags Optionale Werte Optionale Tags zum Filtern und Gruppieren von Zielen auf der Seite „Manage Destinations“ von Cribl Stream. Verwenden Sie eine Registerkarte oder einen harten Zeilenumbruch zwischen Tagnamen. Diese Tags werden verarbeiteten Ereignissen nicht hinzugefügt.

Wählen Sie zum Abschluss Next aus.

Authentifizierungseinstellungen

Wählen Sie auf der Randleiste Authentication Settings aus. Verwenden Sie die Werte, die Sie unter Erstellen eines Microsoft Entra-Dienstprinzipals gespeichert haben, zusammen mit Ihrem Basis-URI wie folgt:

Einstellung Wert Beschreibung
Mandanten-ID <TenantID> Verwenden Sie den tenant-Wert, den Sie unter Erstellen eines Microsoft Entra-Dienstprinzipals gespeichert haben.
Client ID <ClientID> Verwenden Sie die appId-Werte, die Sie unter Erstellen eines Microsoft Entra-Dienstprinzipals gespeichert haben.
Umfang <baseuri>/.default Verwenden Sie für baseuri den Wert aus Basis-URI.
Authentifizierungsmethode Geheimer Clientschlüssel, Geheimer Clientschlüssel (Textgeheimnis) oder Zertifikat Zu den Optionen gehört Geheimer Clientschlüssel. Verwenden Sie den geheimen Clientschlüssel der Microsoft Entra-Anwendung, die Sie unter Erstellen eines Microsoft Entra-Dienstprinzipals für Geheimer Clientschlüssel erstellt haben. Für Zertifikat verwendet Ihr Zertifikat den öffentlichen Schlüssel, den Sie für die unter Erstellen eines Microsoft Entra-Dienstprinzipals erstellte Microsoft Entra-Anwendung registriert haben/registrieren werden.

Wählen Sie Weiteraus.

Persistente Warteschlange

Wird angezeigt, wenn Ingestion mode auf Streaming und Backpressure behavior auf Persistent Queue festgelegt ist.

Einstellung Wert Beschreibung
Max file size 1 MB (Standard) Die maximale Dateigröße der Warteschlange vor dem Schließen der Datei. Beziehen Sie Einheiten wie KB oder MB ein, wenn Sie eine Zahl eingeben.
Max queue size 5 GB (Standard) Der maximale Speicherplatz, den die Warteschlange für jeden Workerprozess verbrauchen kann, bevor das Ziel aufhört, die Daten in die Warteschlange einzureihen. Ein erforderlicher Wert positiver Zahlen mit Einheiten wie KB, MB oder GB. Der Höchstwert ist 1 TB.
Queue file path $CRIBL_HOME/state/queues (Standard) Der Speicherort der persistenten Warteschlangendatei. Cribl Stream fügt /<worker‑id>/<output‑id> an diesen Wert an.
Komprimierung None (Standard), gzip Die Komprimierungsmethode, die zum Komprimieren der persistenten Daten beim Schließen verwendet werden soll
Queue-full behavior Block oder Drop Wählen Sie aus, ob Ereignisse blockiert oder weggelassen werden sollen, wenn die Warteschlange aufgrund geringer Datenträgerkapazität oder vollständiger Datenträgerauslastung einen Rückstau aufweist.
Strict ordering Yes (Standard) oder No Bei Festlegen auf Yes werden Ereignisse basierend auf der Sortierung „First In, First Out“ weitergeleitet. Legen Sie die Option auf No fest, um neue Ereignisse vor früheren in die Warteschlange eingereihten Ereignissen zu senden.
Drain rate limit (EPS) 0 (Standard) Diese Option wird angezeigt, wenn Strict ordering auf No festgelegt ist, damit Sie eine Einschränkungsrate (in Ereignissen pro Sekunde) für das Schreiben aus der Warteschlange in Empfänger festlegen können. Durch die Einschränkung der Ausgleichsrate von Ereignissen in der Warteschlange wird der neue oder aktive Verbindungsdurchsatz erhöht. Bei „Null“ wird die Einschränkung deaktiviert.
Clear Persistent Queue Nicht verfügbar Wählen Sie diese Option aus, um Dateien zu löschen, die aktuell zur Übermittlung an Ihr Ziel in die Warteschlange eingereiht sind. Sie müssen diese Aktion bestätigen, da in die Warteschlange eingereihte Daten dauerhaft gelöscht werden, ohne zugestellt zu werden.

Wählen Sie nach Abschluss der Einstellungen Weiter aus.

Verarbeitungseinstellungen

Einstellung Wert Beschreibung
Pipeline <\defined_pipeline> Eine optionale Pipeline zum Verarbeiten von Daten vor dem Senden mithilfe dieser Ausgabe
System fields cribl_pipe (Standard), cribl_host, cribl_input, cribl_output, cribl_route oder cribl_wp Eine Liste von Feldern, die Ereignissen automatisch hinzugefügt werden, bevor sie an ihr Ziel gesendet werden. Platzhalter werden unterstützt.

Wählen Sie nach Abschluss der Einstellungen Weiter aus.

Parquet-Einstellungen

Wird angezeigt, wenn Parquet für Data Formatausgewählt ist.

Durch Auswählen von Parquet wird die Registerkarte „Parquet Settings“ geöffnet, um das Parquet-Schema auszuwählen.

Einstellung Wert Beschreibung
Automatic schema Ein oder Aus Wählen Sie On aus, um ein Parquet-Schema basierend auf den Ereignissen jeder Parquet-Datei zu generieren, die Cribl Stream schreibt.
Parquet schema Dropdown Wird angezeigt, wenn Automatic schema auf Off festgelegt ist, damit Sie Ihr Parquet-Schema auswählen können.
Parquet version 1.0, 2.4, 2.6 (Standard) Die Version bestimmt die unterstützten Datentypen und deren Darstellung.
Data page version V1, V2 (Standard) Das Serialisierungsformat der Datenseite. Wenn Ihr Parquet-Leser Parquet V2 nicht unterstützt, verwenden Sie V1.
Group row limit 1.000 (Standard) Die maximale Anzahl von Zeilen, die jede Gruppe enthalten kann
Seitengröße 1 MB (Standard) Die Zielspeichergröße für Seitensegmente. Niedrigere Werte können die Lesegeschwindigkeit, höhere Werte die Komprimierung verbessern.
Log invalid rows Ja oder Nein Wenn Yes ausgewählt ist und Log level auf debug festgelegt ist, werden bis zu 20 eindeutige Zeilen ausgegeben, die aufgrund einer fehlenden Datenformatübereinstimmung übersprungen wurden.
Write statistics On (Standard) oder Off Wählen Sie On aus, wenn die Anzeigetools für die Parquet-Statistik konfiguriert sind.
Write page indexes On (Standard) oder Off Wählen Sie On aus, wenn Ihr Parquet-Leser das Überspringen von Seiten mit der Statistik des Parquet-Seitenindexes aktiviert.
Write page checksum Ein oder Aus Wählen Sie Ein aus, wenn Sie Parquet-Tools verwenden, um die Datenintegrität mithilfe von Prüfsummen von Parquet-Seiten zu überprüfen.
Metadata (optional)* Die Metadateneigenschaften der Zieldatei, die als Schlüsselwertpaare aufgenommen werden können.

Wiederholungsversuche

Wird angezeigt, wenn der Erfassungsmodus auf Streaming festgelegt ist.

Einstellung Wert Beschreibung
Honor Retry-After header Ja oder Nein Option zur Berücksichtigung des Retry-After-Headers. Wenn diese Option aktiviert ist, hat ein empfangener Retry-After-Header Vorrang und wird vor anderen konfigurierten Optionen im Abschnitt „Wiederholungsversuche“ verwendet, solange der Header eine Verzögerung von höchstens 180 Sekunden angibt. Andernfalls werden Retry-After-Header ignoriert.
Settings for failed HTTP requests HTTP-Statuscodes (Azure Cognitive Search) Eine Liste der HTTP-Statuscodes für die automatische Wiederholung, wenn keine Verbindung hergestellt wird. Cribl Stream führt automatisch Wiederholungsversuche für Anforderungen mit Fehlercode 429 durch.
Retry timed-out HTTP requests Ein oder Aus Wird diese Option aktiviert, werden weitere Einstellungen für Wiederholungsverhalten verfügbar.
Pre-backoff interval (ms) 1.000 ms (Standard) Die Wartezeit vor dem Wiederholungsversuch
Backoff multiplier 2 s (Standard) Wird als Basis für den Algorithmus des exponentiellen Backoff verwendet, um das Intervall zwischen Wiederholungen zu bestimmen.
Backoff limit (ms) 10.000 ms (Standard) Das maximale Backoffintervall für den endgültigen Streaming-Wiederholungsversuch. Mögliche Werte reichen von 10.000 Millisekunden (10 Sekunden) bis 180.000 Millisekunden (3 Minuten).

Wählen Sie nach Abschluss der Einstellungen Weiter aus.

Erweiterte Einstellungen

Wählen Sie auf der Randleiste die Option Advanced Settings aus. Im Folgenden werden die erweiterten Einstellungen beschrieben, wenn Batching ausgewählt ist:

Einstellung Wert Beschreibung
Flush immediately Yes oder No (Standard) Auf Yes festgelegt, um die Datenaggregation in Kusto außer Kraft zu setzen. Weitere Informationen finden Sie unter Bewährte Methoden für die Kusto-Erfassungsbibliothek.
Retain blob on success Yes oder No (Standard) Wählen Sie Yes aus, um das Datenblob nach der Erfassung beizubehalten.
Extent tags <\ExtentTag, ET2,...> Legen Sie bei Bedarf Tags auf partitionierte Erweiterungen der Zieltabelle fest.
Enforce uniqueness via tag values Wählen Sie Add value aus, um eine ingest-by-Werteliste anzugeben, die zum Filtern eingehender Erweiterungen verwendet werden soll, und verwerfen Sie die Erweiterungen, die einem aufgelisteten Wert entsprechen. Weitere Informationen finden Sie unter Erweiterungen (Datenshards).
Report level DoNotReport, FailuresOnly (Standard) und FailuresAndSuccesses Die Statusberichtebene der Erfassung
Report method Queue (Standard), Tableund QueueAndTable (empfohlen) Ziel für den Erfassungsstatusbericht
Zusätzliche Felder Fügen Sie bei Bedarf weitere Konfigurationseigenschaften hinzu, die an den Erfassungsdienst gesendet werden sollen.
Staging location /tmp (Standard) Lokaler Dateisystemspeicherort, an dem Dateien vor dem Komprimieren und Verschieben in das endgültige Ziel gepuffert werden sollen. Cribl empfiehlt einen stabilen und leistungsfähigen Speicherort.
File name suffix expression .${C.env["CRIBL_WORKER_ID"]}.${__format}${__compression === "gzip" ? ".gz" : ""}(Standard) Ein JavaScript-Ausdruck in Anführungszeichen oder Backticks, die als Suffix des Ausgabedateinamens verwendet werden. format ist JSON oder Raw, und __compression ist Keine oder gzip. Am Ende der Dateinamen wird eine zufällige sechsstellige Zeichenfolge angehängt, um zu verhindern, dass die Dateien überschrieben werden.
Max file size (MB) 32 MB (Standard) Die maximale unkomprimierte Ausgabedateigröße von Dateien, bevor diese geschlossen und in den Speichercontainer verschoben werden
Max file open time (sec) 300 Sekunden (Standard) Die maximale Zeit in Sekunden, um in eine Datei zu schreiben, bevor diese geschlossen und in den Speichercontainer verschoben wird
Max file idle time (sec) 30 Sekunden (Standard) Die maximale Zeit in Sekunden, um inaktive Dateien geöffnet zu lassen, bevor diese geschlossen und in den Speichercontainer verschoben werden
Max open files 100 (Standard) Die maximale Anzahl von Dateien, die gleichzeitig geöffnet sind, bevor die ältesten geöffneten Dateien geschlossen und in den Speichercontainer verschoben werden
Max concurrent file parts 1 (Standard) Die maximale Anzahl von Dateiteilen, die gleichzeitig hochgeladen werden sollen. Der Standardwert ist 1, und der höchste Wert ist 10. Durch Festlegen des Werts auf 1 werden die Teile einzeln nacheinander gesendet.
Remove empty staging dirs Yes (Standard) oder No Bei Aktivierung löscht Cribl Stream leere Stagingverzeichnisse nach dem Verschieben von Dateien. Dadurch werden verwaiste leere Verzeichnisse vermieden. Bei Aktivierung wird der Staging cleanup period angezeigt.
Staging cleanup period 300 (Standard) Der Zeitraum in Sekunden, bis leere Verzeichnisse gelöscht werden, wenn Remove staging dirs aktiviert ist. Wird angezeigt, wenn Remove empty staging dirs auf Yes festgelegt ist. Der Mindestwert beträgt 10, der Höchstwert 86.400 Sekunden (alle 24 Stunden).
Umgebung Wird die Option leer gelassen (Standardeinstellung), ist die Konfiguration überall aktiviert. Bei Verwendung von GitOps geben Sie den Git-Branch an, in dem Sie die Konfiguration aktivieren möchten.

Wählen Sie danach Save aus.

Verbindungskonfiguration

Wählen Sie im daraufhin geöffneten Fenster Connection Configuration die Option Passthru connection und dann Save aus. Der Connector fängt an, die Daten in die Warteschlange einzureihen.

Bestätigen der Datenerfassung

  1. Sobald die Daten in der Tabelle erscheinen, bestätigen Sie durch Überprüfen der Zeilenanzahl die Datenübertragung:

    <Tablename> 
| count

  2. Bestätigen Sie die Erfassungen, die in den letzten fünf Minuten in die Warteschlange eingereiht wurden:

    .show commands-and-queries 
| where Database == "" and CommandType == "DataIngestPull" 
| where LastUpdatedOn >= ago(5m)

  3. Vergewissern Sie sich, dass der Erfassungsprozess keine Fehler aufweist:

    • Für Batchverarbeitung:
    .show ingestion failures
    • Für Streaming:
    .show streamingingestion failures 
| order by LastFailureOn desc

  4. Überprüfen Sie Daten in Ihrer Tabelle:

    <TableName>
| take 10

Abfragebeispiele und einen Abfrageleitfaden finden Sie unter Schreiben von Abfragen in KQL und unter Dokumentation zur Kusto-Abfragesprache.

Zugehöriger Inhalt