Pozyskiwanie danych ze strumienia Cribl do usługi Azure Data Explorer
Cribl Stream to aparat przetwarzania, który bezpiecznie zbiera, przetwarza i przesyła strumieniowo dane zdarzeń maszyny z dowolnego źródła. Umożliwia analizowanie i przetwarzanie tych danych dla dowolnego miejsca docelowego na potrzeby analizy i zarządzania w bezpieczny sposób.
W tym artykule pokazano, jak pozyskiwać dane za pomocą usługi Cribl Stream.
Aby uzyskać pełną listę łączników danych, zobacz Omówienie integracji danych.
Wymagania wstępne
- Konto usługi Cribl Stream
- Klaster i baza danych usługi Azure Data Explorer z domyślną pamięcią podręczną i zasadami przechowywania.
- Środowisko zapytań. Aby uzyskać więcej informacji, zobacz Omówienie integracji zapytań.
- Identyfikator URI klastra Kusto dla wartości TargetURI w klastrze> formatuhttps://ingest-<.<region.kusto.windows.net>. Aby uzyskać więcej informacji, zobacz Dodawanie połączenia klastra.
Tworzenie jednostki usługi Entra firmy Microsoft
Jednostkę usługi Microsoft Entra można utworzyć za pośrednictwem witryny Azure Portal lub programowo, jak w poniższym przykładzie.
Ta jednostka usługi jest tożsamością używaną przez łącznik do zapisywania danych w tabeli w usłudze Kusto. Przyznasz uprawnienia dla tej jednostki usługi w celu uzyskania dostępu do zasobów usługi Kusto.
Zaloguj się do subskrypcji platformy Azure za pomocą interfejsu wiersza polecenia platformy Azure. Następnie uwierzytelnij się w przeglądarce.
az loginWybierz subskrypcję do hostowania podmiotu zabezpieczeń. Ten krok jest wymagany, gdy masz wiele subskrypcji.
az account set --subscription YOUR_SUBSCRIPTION_GUIDUtwórz jednostkę usługi. W tym przykładzie jednostka usługi nosi nazwę
my-service-principal.az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}Z zwróconych danych JSON skopiuj wartości
appId,passworditenantdo użycia w przyszłości.{ "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" }
Utworzono aplikację Microsoft Entra i jednostkę usługi.
Tworzenie tabeli docelowej
Utwórz tabelę docelową dla danych przychodzących i mapowanie pozyskiwania w celu zamapowania pozyskanych kolumn danych na kolumny w tabeli docelowej.
Uruchom następujące polecenie tworzenia tabeli w edytorze zapytań, zastępując symbol zastępczy TableName nazwą tabeli docelowej:
.create table <TableName> (_raw: string, _time: long, cribl_pipe: dynamic)Uruchom następujące polecenie tworzenia mapowania pozyskiwania, zastępując symbole zastępcze TableName nazwą tabeli docelowej i TableNameMapping nazwą mapowania pozyskiwania:
.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}]'Udziel jednostce usługi uprawnienia do pracy z bazą danych w bazie danych utwórz jednostkę usługi Microsoft Entra. Aby uzyskać więcej informacji, zobacz Przykłady. Zastąp symbol zastępczy DatabaseName nazwą docelowej bazy danych i identyfikatorem ApplicationID wartością
AppIdzapisaną podczas tworzenia jednostki usługi Microsoft Entra..add database <DatabaseName> ingestors ('aadapp=<ApplicationID>') 'App Registration'
Tworzenie miejsca docelowego usługi Cribl Stream
W poniższej sekcji opisano sposób tworzenia miejsca docelowego usługi Cribl Stream, który zapisuje dane w tabeli w usłudze Kusto. Każda tabela wymaga oddzielnego łącznika docelowego usługi Cribl Stream.
Wybieranie miejsca docelowego
Aby połączyć usługę Cribl Stream z tabelą:
Z górnej nawigacji w cribl wybierz pozycję Zarządzaj , a następnie wybierz grupę procesów roboczych.
Wybierz pozycję Routing>QuickConnect (Strumień)>Dodaj lokalizację docelową.
W oknie Konfigurowanie nowego miejsca docelowego szybkiego połączenia wybierz pozycję Azure Data Explorer, a następnie pozycję Dodaj teraz.
Konfigurowanie ustawień ogólnych
W oknie Nowy Eksplorator danych w obszarze Ustawienia ogólne ustaw następujące ustawienia:
| Ustawienie | Wartość | Opis |
|---|---|---|
| Identyfikator danych wyjściowych | <OutputID>, na przykład KustoDestination | Nazwa używana do identyfikowania miejsca docelowego. |
| Tryb pozyskiwania | Przetwarzanie wsadowe (domyślne) lub przesyłanie strumieniowe | Ustawienia trybu pozyskiwania. Przetwarzanie wsadowe umożliwia ściąganie partii danych z kontenera magazynu Cribl podczas pozyskiwania dużych ilości danych w krótkim czasie. Przesyłanie strumieniowe wysyła dane bezpośrednio do docelowej tabeli KQL. Przesyłanie strumieniowe jest przydatne do pozyskiwania mniejszych ilości danych lub na przykład wysyłania alertu krytycznego w czasie rzeczywistym. Przesyłanie strumieniowe może osiągnąć mniejsze opóźnienie niż przetwarzanie wsadowe. Jeśli tryb pozyskiwania jest ustawiony na Przesyłanie strumieniowe, musisz włączyć zasady przesyłania strumieniowego. Aby uzyskać więcej informacji, zobacz Zasady pozyskiwania przesyłania strumieniowego. |
| Identyfikator URI podstawowy klastra | podstawowy identyfikator URI | Podstawowy identyfikator URI. |
| Identyfikator URI usługi pozyskiwania | identyfikator URI pozyskiwania | Wyświetla, gdy wybrano tryb wsadowania . Identyfikator URI pozyskiwania. |
| Nazwa bazy danych | <Nazwa bazy danych> | Nazwa docelowej bazy danych. |
| Nazwa tabeli | <TableName> | Nazwa tabeli docelowej. |
| Weryfikowanie ustawień bazy danych | Tak (ustawienie domyślne) lub Nie. | Sprawdza poprawność poświadczeń aplikacji jednostki usługi wprowadzonych podczas zapisywania lub uruchamiania miejsca docelowego. Sprawdza poprawność nazwy tabeli, z wyjątkiem sytuacji, w których jest włączony obiekt dodawania mapowania. To ustawienie powinno być wyłączone, jeśli aplikacja nie ma ról Przeglądarki baz danych i Przeglądarki tabel. |
| Dodawanie obiektu mapowania | Tak lub Nie (wartość domyślna). | Wyświetlane tylko w przypadku wybrania trybu batching zamiast domyślnego pola tekstowego mapowania danych. Wybranie pozycji Tak powoduje otwarcie okna w celu wprowadzenia mapowania danych jako obiektu JSON. |
| Mapowanie danych | Nazwa schematu mapowania zgodnie z definicją w kroku Tworzenie tabeli docelowej. | Nazwa schematu mapowania. Widok domyślny, gdy dla opcji Dodaj obiekt mapowania ustawiono wartość Nie. |
| Kompresować | gzip (ustawienie domyślne) | Gdy format danych jest ustawiony na Parquet, kompresja nie jest dostępna. |
| Format danych | JSON (ustawienie domyślne), Nieprzetworzone lub Parquet. | Format danych. Parquet jest dostępny tylko w trybie batching i jest obsługiwany tylko w systemie Linux. |
| Zachowanie wsteczne | Blokuj (wartość domyślna) lub Upuść | Wybierz, czy mają być blokowane czy upuszczające zdarzenia, gdy odbiorniki wywierają ciśnienie wsteczne. |
| Tagi | Wartości opcjonalne | Opcjonalne tagi do filtrowania i grupowania miejsc docelowych na stronie Zarządzanie miejscami docelowymi usługi Cribl Stream. Użyj karty lub twardego powrotu między nazwami tagów. Te tagi nie są dodawane do przetworzonych zdarzeń. |
Po zakończeniu wybierz pozycję Dalej.
Ustawienia uwierzytelniania
Wybierz pozycję Ustawienia uwierzytelniania na pasku bocznym. Użyj wartości zapisanych w sekcji Tworzenie jednostki usługi Microsoft Entra wraz z podstawowym identyfikatorem URI w następujący sposób:
| Ustawienie | Wartość | Opis |
|---|---|---|
| Identyfikator dzierżawy | <Identyfikator dzierżawy> | Użyj wartości zapisanej tenant w temacie Tworzenie jednostki usługi Microsoft Entra. |
| Client ID | <ClientID> | Użyj wartości zapisanych appId w temacie Tworzenie jednostki usługi Entra firmy Microsoft. |
| Scope | <baseuri>/.default |
Użyj wartości z podstawowego identyfikatora URI dla identyfikatora baseuri. |
| Metoda uwierzytelniania | Klucz tajny klienta, klucz tajny klienta (wpis tajny) lub certyfikat | Opcje to Klucz tajny klienta Użyj wpisu tajnego klienta aplikacji Microsoft Entra utworzonej w sekcji Tworzenie jednostki usługi Entra firmy Microsoft dla wpisu tajnego klienta. W przypadku certyfikatu certyfikat używa zarejestrowanego/zarejestrowanego klucza publicznego dla aplikacji Microsoft Entra utworzonej w sekcji Tworzenie jednostki usługi Microsoft Entra. |
Następnie kliknij przycisk Dalej.
Kolejka trwała
Wyświetla, gdy tryb pozyskiwania jest ustawiony na wartość Przesyłanie strumieniowe, a zachowanie Backpressure jest ustawione na kolejkę trwałą.
| Ustawienie | Wartość | Opis |
|---|---|---|
| Maksymalny rozmiar pliku | 1 MB (wartość domyślna) | Maksymalny rozmiar pliku kolejki do osiągnięcia przed zamknięciem pliku. Uwzględnij jednostki, takie jak KB lub MB, podczas wprowadzania numeru. |
| Maksymalny rozmiar kolejki | 5 GB (ustawienie domyślne) | Maksymalna ilość miejsca na dysku, które kolejka może zużywać w każdym procesie roboczym przed zatrzymaniem kolejkowania danych przez miejsce docelowe. Wymagana wartość liczb dodatnich z jednostkami, takimi jak KB, MB lub GB. Maksymalna wartość to 1 TB. |
| Ścieżka pliku kolejki | $CRIBL_HOME/state/queues (domyślne) |
Lokalizacja pliku trwałej kolejki. Usługa Cribl Stream dołącza /<worker‑id>/<output‑id> do tej wartości. |
| Kompresja | Brak (wartość domyślna), gzip | Metoda kompresji używana do kompresowania utrwalone dane po zamknięciu. |
| Zachowanie pełne kolejki | Blokuj lub upuść | Wybierz opcję blokowania lub upuszczania zdarzeń, gdy kolejka wywiera presję wsteczną z powodu małej pojemności dysku lub pełnej pojemności dysku. |
| Ścisłe porządkowanie | Tak (ustawienie domyślne) lub Nie | Po ustawieniu wartości Tak zdarzenia są przekazywane na podstawie pierwszego w, najpierw kolejność na wyjęciu. Ustaw wartość Nie , aby wysyłać nowe zdarzenia przed wcześniejszymi zdarzeniami w kolejce. |
| Limit szybkości opróżniania (EPS) | 0 (wartość domyślna) | Ta opcja jest wyświetlana, gdy ustawienie Ścisłe kolejność ma wartość Nie, aby umożliwić ustawienie szybkości ograniczania (w zdarzeniach na sekundę) na zapis z kolejki do odbiorników. Ograniczanie szybkości opróżniania zdarzeń w kolejce zwiększa przepływność nowych lub aktywnych połączeń. Zero wyłącza ograniczanie przepustowości. |
| Wyczyść kolejkę trwałą | NA | Wybierz, aby usunąć pliki aktualnie w kolejce do dostarczenia do miejsca docelowego. Musisz potwierdzić tę akcję, ponieważ dane w kolejce są trwale usuwane bez dostarczania. |
Po zakończeniu wybierz przycisk Dalej.
Ustawienia przetwarzania
| Ustawienie | Wartość | Opis |
|---|---|---|
| Rurociąg | <\defined_pipeline> | Opcjonalny potok do przetwarzania danych przed wysłaniem ich przy użyciu tych danych wyjściowych. |
| Pola systemowe | cribl_pipe (ustawienie domyślne), cribl_host, cribl_input, cribl_output, cribl_routelub cribl_wp |
Lista pól, które są automatycznie dodawane do zdarzeń przed ich wysłaniem do miejsca docelowego. Obsługiwane są symbole wieloznaczne. |
Po zakończeniu wybierz przycisk Dalej.
Ustawienia parquet
Wyświetla, gdy parquet jest wybrany dla formatu danych.
Wybranie pozycji Parquet powoduje otwarcie karty Ustawienia parquet, aby wybrać schemat Parquet.
| Ustawienie | Wartość | Opis |
|---|---|---|
| Schemat automatyczny | Włączone lub wyłączone | Wybierz pozycję Wł. , aby wygenerować schemat Parquet na podstawie zdarzeń każdego pliku Parquet zapisywanego przez usługę Cribl Stream. |
| Schemat Parquet | lista rozwijana | Wyświetla, gdy dla schematu automatycznego jest ustawiona wartość Wyłączone , aby umożliwić wybranie schematu parquet. |
| Wersja Parquet | 1.0, 2.4, 2.6 (ustawienie domyślne) | Wersja określa obsługiwane typy danych i sposób ich reprezentowania. |
| Wersja strony danych | Wersja 1, wersja 2 (ustawienie domyślne) | Format serializacji strony danych. Jeśli czytnik Parquet nie obsługuje parquet v2, użyj wersji 1. |
| Limit wierszy grupy | 1000 (ustawienie domyślne) | Maksymalna liczba wierszy, które mogą zawierać każda grupa. |
| Rozmiar strony | 1 MB (wartość domyślna) | Docelowy rozmiar pamięci dla segmentów stron. Niższe wartości mogą poprawić szybkość odczytu, podczas gdy wyższe wartości mogą poprawić kompresję. |
| Rejestrowanie nieprawidłowych wierszy | Tak lub Nie | Gdy wybrano opcję Tak , a poziom dziennika jest ustawiony na debugwartość , zwraca maksymalnie 20 unikatowych wierszy, które zostały pominięte z powodu niezgodności formatu danych. |
| Pisanie statystyk | Włączone (domyślne) lub Wyłączone | Wybierz pozycję Wł. , jeśli masz skonfigurowane narzędzia do wyświetlania statystyk Parquet. |
| Zapisywanie indeksów stron | Włączone (domyślne) lub Wyłączone | Wybierz pozycję Wł. , jeśli czytnik Parquet używa statystyk indeksu stron Parquet, aby włączyć pomijanie strony. |
| Pisanie sumy kontrolnej strony | Włączone lub wyłączone | Wybierz pozycję Wł. , jeśli używasz narzędzi Parquet do sprawdzania integralności danych przy użyciu sum kontrolnych stron Parquet. |
| Metadane (opcjonalne)* | Właściwości metadanych pliku docelowego, które można uwzględnić jako pary klucz-wartość. |
Ponowne próby
Wyświetla, gdy tryb pozyskiwania jest ustawiony na Wartość Przesyłanie strumieniowe.
| Ustawienie | Wartość | Opis |
|---|---|---|
| Honor Retry-After nagłówek | Tak lub Nie | Czy należy uhonorować Retry-After nagłówek. Po włączeniu odebrany Retry-After nagłówek ma pierwszeństwo przed innymi skonfigurowanymi opcjami w sekcji Ponawianie prób, o ile nagłówek określa opóźnienie 180 sekund lub mniej. Retry-After W przeciwnym razie nagłówki są ignorowane. |
| Ustawienia żądań HTTP, które zakończyły się niepowodzeniem | Kody stanu HTTP | Lista kodów stanu HTTP, które mają być automatycznie ponawiane, jeśli nie uda im się nawiązać połączenia. Usługa Cribl Stream automatycznie ponawia próby 429 nieudanych żądań. |
| Ponawianie próby przekroczenia limitu czasu żądań HTTP | Włączone lub wyłączone | Po ustawieniu stają się dostępne więcej ustawień zachowania ponawiania prób. |
| Interwał przed wycofywaniem (ms) | 1000 ms (ustawienie domyślne) | Czas oczekiwania przed ponowną próbą. |
| Mnożnik wycofywania | 2 s (ustawienie domyślne) | Służy jako podstawa algorytmu wykładniczego wycofywania w celu określenia interwału między ponawiania próbami. |
| Limit wycofywania (ms) | 10 000 ms (wartość domyślna) | Maksymalny interwał wycofywania końcowego ponawiania próby przesyłania strumieniowego. Możliwe wartości wahają się od 10 000 milisekund (10 sekund) do 180 000 milisekund (3 minuty). |
Po zakończeniu wybierz przycisk Dalej.
Ustawienia zaawansowane
Wybierz pozycję Ustawienia zaawansowane na pasku bocznym. Poniżej opisano ustawienia zaawansowane podczas wybierania usługi Batching :
| Ustawienie | Wartość | Opis |
|---|---|---|
| Opróżnij natychmiast | Tak lub Nie (wartość domyślna). | Ustaw wartość Tak , aby zastąpić agregację danych w usłudze Kusto. Aby uzyskać więcej informacji, zobacz Best practices for the Kusto Ingest library (Najlepsze rozwiązania dotyczące biblioteki pozyskiwania Kusto). |
| Zachowaj obiekt blob przy powodzeniu | Tak lub Nie (wartość domyślna). | Ustaw wartość Tak , aby zachować obiekt blob danych po zakończeniu pozyskiwania. |
| Tagi zakresu | <\ExtentTag, ET2,...> | Ustaw tagi, jeśli to konieczne, na partycjonowane zakresy tabeli docelowej. |
| Wymuszanie unikatowości za pomocą wartości tagów | Wybierz pozycję Dodaj wartość , aby określić ingest-by listę wartości, która ma być używana do filtrowania zakresów przychodzących i odrzucania zakresów pasujących do wartości wymienionej. Aby uzyskać więcej informacji, zobacz Zakresy (fragmenty danych) |
|
| Poziom raportu | DoNotReport, FailuresOnly (wartość domyślna) i FailuresAndSuccesses. | Poziom raportowania stanu pozyskiwania. |
| Metoda raportu | Queue (ustawienie domyślne), Table i QueueAndTable (zalecane). | Cel raportowania stanu pozyskiwania. |
| Dodatkowe pola | Dodaj więcej właściwości konfiguracji, jeśli to konieczne, aby wysłać do usługi pozyskiwania. | |
| Lokalizacja przemieszczania | /tmp (domyślne) |
Lokalna lokalizacja systemu plików, w której mają być buforowane pliki przed skompresowaniem i przeniesieniem ich do końcowego miejsca docelowego. Cribl zaleca stabilną i wysokowydajną lokalizację. |
| Wyrażenie sufiksu nazwy pliku | .${C.env["CRIBL_WORKER_ID"]}.${__format}${__compression === "gzip" ? ".gz" : ""}(ustawienie domyślne) |
Wyrażenie języka JavaScript ujęte w cudzysłowy lub backticks używane jako sufiks nazwy pliku wyjściowego. format może być w formacie JSON lub nieprzetworzonym i __compression może być żaden lub gzip. Losowa sekwencja sześciu znaków jest dołączana na końcu nazw plików, aby zapobiec ich zastępowaniu. |
| Maksymalny rozmiar pliku (MB) | 32 MB (ustawienie domyślne) | Maksymalny nieskompresowany rozmiar pliku wyjściowego, który można osiągnąć przed ich zamknięciem i zostanie przeniesiony do kontenera magazynu. |
| Maksymalny czas otwierania pliku (s) | 300 sekund (wartość domyślna) | Maksymalny czas zapisu w pliku w sekundach przed jego zamknięciem i przeniesieniem do kontenera magazynu. |
| Maksymalny czas bezczynności pliku (s) | 30 sekund (ustawienie domyślne) | Maksymalny czas (w sekundach) przechowywania nieaktywnych plików przed ich zamknięciem i jest przenoszony do kontenera magazynu. |
| Maksymalna liczba otwartych plików | 100 (ustawienie domyślne) | Maksymalna liczba plików do otwarcia w tym samym czasie przed zamknięciem najstarszych otwartych plików i przeniesieniem ich do kontenera magazynu. |
| Maksymalna liczba współbieżnych części plików | 1 (ustawienie domyślne) | Maksymalna liczba części plików do przekazania w tym samym czasie. Wartość domyślna to 1, a najwyższa wartość to 10. Ustawienie wartości na jedną umożliwia sekwencyjne wysyłanie jednej części. |
| Usuń puste różnice przejściowe | Tak (ustawienie domyślne) lub Nie | Po przełączeniu w usłudze Cribl Stream puste katalogi przejściowe są usuwane po przeniesieniu plików. Zapobiega to rozprzestrzenianiu się oddzielonych pustych katalogów. Po włączeniu program uwidacznia okres oczyszczania przejściowego. |
| Okres oczyszczania przejściowego | 300 (ustawienie domyślne) | Ilość czasu w sekundach do usunięcia pustych katalogów po włączeniu usuwania tymczasowych dirs . Wyświetla, gdy opcja Usuń puste różnice przejściowe jest ustawiona na Wartość Tak. Wartość minimalna to 10 sekund, a wartość maksymalna to 86 400 sekund (co 24 godziny). |
| Środowisko | Gdy pusta (domyślna) konfiguracja jest włączona wszędzie. Jeśli używasz metodyki GitOps, możesz określić gałąź Git, w której chcesz włączyć konfigurację. |
Po zakończeniu wybierz pozycję Zapisz.
Konfiguracja połączenia
W oknie Konfiguracja połączenia, które zostanie otwarte, wybierz pozycję Połączenie passthru, a następnie pozycję Zapisz. Łącznik rozpoczyna kolejkowanie danych.
Potwierdzanie pozyskiwania danych
Po nadejściu danych w tabeli potwierdź transfer danych, sprawdzając liczbę wierszy:
<Tablename> | countPotwierdź pozyskiwanie w kolejce w ciągu ostatnich pięciu minut:
.show commands-and-queries | where Database == "" and CommandType == "DataIngestPull" | where LastUpdatedOn >= ago(5m)Upewnij się, że w procesie pozyskiwania nie występują żadne błędy:
- W przypadku przetwarzania wsadowego:
.show ingestion failures- W przypadku przesyłania strumieniowego:
.show streamingingestion failures | order by LastFailureOn descSprawdź dane w tabeli:
<TableName> | take 10
Aby uzyskać przykłady zapytań i wskazówki, zobacz Pisanie zapytań w języku KQL i język zapytań Kusto dokumentacji.