Czy zastanawiał[a|e]ś się kiedyś co zrobić, żeby każdy nowo dodany do Elasticsearcha dokument zawierał datę jego dodania bez konieczności przesyłania jej z zewnątrz?
Do wersji 5.0 było to całkiem proste. Wystarczyło skorzystać ze specjalnego pola _timestamp. Jednak od wersji 2.0 zaczęto z niego rezygnować.
Oczywiście zawsze można taką datę przesyłać z zewnątrz i wrzucać ją do dowolnie wybranego przez nas pola. Tu pojawia się jednak pewien problem. Mianowicie, jeżeli istnieje więcej niż jedno źródło danych naszego Elasticsearcha, to w każdym z nich musimy powielić to zachowanie. Sytuacja staje się tym bardziej kłopotliwa im więcej mamy owych źródeł.
Czy mamy więc coś po stronie Elasticsearcha, co zastąpiłoby nam zachowanie starego, dobrego _timestamp-a? Mamy! Możemy wykorzystać do tego ingest node ze specjalnie zdefiniowanym pipelinem. Co więcej, ingest node możemy wykorzystać do wielu innych akcji preprocessingu dokumentów. I właśnie o tym chciałbym Ci dzisiaj opowiedzieć.
Ingest node
Domyślnie rolę ingest node’a (jak to dobrze przetłumaczyć na polski? węzeł łykający? 😉 ) przyjmuje każdy węzeł z klastra Elasticsearcha. Dzięki temu każdy z węzłów może zająć się wstępnym procesowaniem dodawanego dokumentu. W każdej chwili możemy zmienić konfigurację klastra i odebrać wybranym węzłom role ingest node’a (o ile tylko mamy ku temu powód).
Pipeline
Żeby przeprocesować dokument definiujemy odpowiedni pipeline, który się tym zajmie. Każdy pipeline składa się z listy procesorów, które wykonywane są jeden po drugim. Do listy dorzucamy jeszcze krótki opis i gotowe.
Podstawowa definicja wygląda więc bardzo prosto:
|
|
{ "description": "...", "processors": [ ... ] } |
API
API do zarządzania pipeline’ami również nie jest skomplikowane. Do dyspozycji dostajemy cztery dedykowane endpointy: do dodawania/aktualizowania, pobierania, usuwania i symulowania działania naszego potoku.
Z kolei wykonanie pipeline’a możemy wywołać na dwa sposoby. Po pierwsze, poprzez wywołanie indeksowania dokumentu (lub operacji bulk) z dodatkowym parametrem pipeline:
|
|
PUT /yoda/_doc/1?pipeline=use_the_source { "quote": "No! Try not! Do or do not, there is no try." } |
Gdzie use_the_source jest identyfikatorem naszego pipeline’a ustalonym podczas jego tworzenia.
Drugim sposobem jest zdefiniowanie domyślnie uruchamianego pipeline’a ustawiając parametr index.default_pipeline w ustawieniach indeksu. Należy jednak pamiętać, że domyślny potok zostanie użyty tylko wtedy, gdy w URL-u nie wskażemy innego.
Procesory
Ogólna definicja procesora wygląda następująco:
|
|
{ "PROCESSOR_NAME": { ... processor configuration options ... } } |
Szczegóły konfiguracyjne zależą od tego z jakiego typu procesora skorzystamy. A jest ich całkiem sporo. W wersji 7.0 naliczyłem prawie trzydzieści. Przyglądając się całej liście muszę przyznać, że naprawdę jest w czym wybierać.
Co ciekawe, oprócz procesorów przeznaczonych do manipulacji polami (jak na przykład dodanie pola, usunięcie, zmianę typu, zmianę nazwy, rozdzielenie pojedynczej wartości w tablicę, połączenie tablicy w pojedynczą wartość, sparsowanie wartości w odpowiednią strukturę) znajdziemy również takie, które pomogą nam dodać indeksowany dokument do zupełnie innego indeksu, kiedy tylko zostanie spełniony określony warunek. Dodatkowo, jest również procesor umożliwiający odpalenie… kolejnego pipeline’a! Madness!
if
Wspólną dla wszystkich opcją konfiguracyjną jest warunkowe wykonanie procesora. Dzięki parametrowi if możemy powiedzieć, czy dany procesor ma się wykonać, czy po prostu przeskoczyć do kolejnego w potoku.
Obsługa błędów
Sporo możliwości daje nam również obsługa błędów, która może być określona na dwóch poziomach – dla całego pipeline’a lub dla poszczególnych procesorów. Wszytko dzięki parametrowi on_failure, który, co ciekawe, przyjmuje… listę procesorów do wykonania.
Jeżeli nie chcemy jakoś szczególnie reagować na błędy w pipelinie, możemy po prostu powiedzieć, żeby były one ignorowane (z wykorzystaniem parametru ignore_failure).
Przykłady
Spójrzmy sobie teraz jak to mniej więcej działa.
Wrzucamy pierwszy pipeline:
|
|
PUT /_ingest/pipeline/use_the_source { "description": "Use the Source, Luke!", "processors": [ { "set": { "field": "source_used", "value": true } } ] } |
a następnie dokument z jego użyciem:
|
|
PUT /yoda/_doc/1?pipeline=use_the_source { "quote": "No! Try not! Do or do not, there is no try." } |
Kiedy zapytamy o nowo dodany dokument (GET /yoda/_doc/1) otrzymamy:
|
|
{ ... "_source" : { "quote" : "No! Try not! Do or do not, there is no try.", "source_used" : true } } |
Działa!
Możemy też zablokować możliwość dodawania dokumentów, które nie zawierają cytatu.
Dodajemy pipeline:
|
|
PUT /_ingest/pipeline/do_not_add_if_invalid { "description": "Do not add document if 'quote' is missing.", "processors": [ { "drop": { "if": "ctx.quote == null" } } ] } |
i próbujemy dodać dokument:
|
|
PUT /yoda/_doc/2?pipeline=do_not_add_if_invalid { "old_quote": "Judge me by my size, do you?" } |
W rezultacie widzimy, że żaden dokument nie został dodany:
|
|
{ ... "result" : "noop", "_shards" : { "total" : 0, "successful" : 0, "failed" : 0 } } |
Co potwierdzi nam również zapytanie o ten dokument (GET /yoda/_doc/2).
Ulepszmy nieco nasz pipeline. Powiedzmy, że spodziewamy się, że z kilku starych źródeł możemy cały czas otrzymywać dokumenty z polem old_quote. My chcemy jednak używać nowego pola quote. Zamiast ignorować tego typu dokumenty na starcie, jak to miało miejsce powyżej, podejmijmy próbę zmiany nazwy pola. Jeżeli wszystko pójdzie dobrze, to w indeksie wyląduje dokument w prawidłowym formacie.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
PUT /_ingest/pipeline/update_old_docs_to_the_new_requirements { "description": "Rename 'old_quote' to 'quote' if exists and add document only if 'quote' exists.", "processors": [ { "rename": { "field": "old_quote", "target_field": "quote", "ignore_failure": true } }, { "drop": { "if": "ctx.quote == null" } } ] } |
Dodajemy dokument:
|
|
PUT /yoda/_doc/2?pipeline=update_old_docs_to_the_new_requirements { "old_quote": "Judge me by my size, do you?" } |
i pytamy o niego (GET /yoda/_doc/2):
|
|
{ ... "_source" : { "quote" : "Judge me by my size, do you?" } } |
Prawda, że fajnie to wygląda?
Na koniec spróbujmy jeszcze zreprodukować wspomnianą na początku funkcjonalność pola _timestamp.
Tworzymy pipeline:
|
|
PUT /_ingest/pipeline/set_timestamp { "description": "Set `timestamp`.", "processors": [ { "set": { "field": "timestamp", "value": "{{_ingest.timestamp}}" } } ] } |
i używając go dodajemy dokument:
|
|
PUT /yoda/_doc/1?pipeline=set_timestamp { "quote": "No! Try not! Do or do not, there is no try." } |
Po chwili, kiedy o niego zapytamy (GET /yoda/_doc/1) otrzymujemy:
|
|
{ ... "_source" : { "quote" : "No! Try not! Do or do not, there is no try.", "timestamp" : "2019-04-13T14:29:43.321063100Z" } } |
Mamy to!
Dobra. Starczy tych przykładów. Na pewno łapiesz już o co chodzi. A tak jak wspomniałem, z obecnie dostępnymi typami procesorów nasze możliwości ogranicza wyłącznie wyobraźnia 🙂
Podsumowanie
Jeżeli kiedykolwiek staniesz przed konieczności wstępnego podpicowania dokumentów przed ich zaindeksowaniem, to w pierwszej kolejności zajrzyj do dokumentacji ingest node’a. Jeśli Twój przypadek nie jest nadmiernie skomplikowany, to istnieje duża szansa, że to pipeline przyjdzie Ci z pomocą. Plusem takiego rozwiązania będzie to, że w żadnym dodatkowym miejscu nie będziesz musiał(a) pamiętać o ręcznym wywołaniu procesowania na każdym nowym dokumencie. Domyślnie ustawiony pipeline zawsze będzie trzymał się indeksu.
Miał[a|e]ś okazję używać ingest node’a produkcyjnie? Jak sprawdziło Ci się takie rozwiązanie? Jakie najoryginalniejsze pipeline’a przyszło Ci przygotowywać? Podziel się tym w komentarzu poniżej. Jeżeli masz jakieś dodatkowe pytania, to również śmiało pisz.
Bądź na bieżąco!
Podobają Ci się treści publikowane na moim blogu? Nie chcesz niczego pominąć? Zachęcam Cię do subskrybowania kanału RSS, polubienia fanpage na Facebooku, zapisania się na listę mailingową:
lub śledzenia mnie na Twitterze. Generalnie polecam wykonanie wszystkich tych czynności, bo często zdarza się tak, że daną treść wrzucam tylko w jedno miejsce. Zawsze możesz zrobić to na próbę, a jeśli Ci się nie spodoba – zrezygnować 
Dołącz do grup na Facebooku
Chcesz więcej? W takim razie zapraszam Cię do dołączenia do powiązanych grup na Facebooku, gdzie znajdziesz dodatkowe informacje na poruszane tutaj tematy, możesz podzielić się własnymi doświadczeniami i przemyśleniami, a przede wszystkim poznasz ludzi interesujących się tą samą tematyką co Ty.
W grupie Programista Na Swoim znajdziesz wiele doświadczonych osób chętnych do porozmawiania na tematy krążące wokół samozatrudnienia i prowadzenia programistycznej działalności gospodarczej. Vademecum Juniora przeznaczone jest zaś do wymiany wiedzy i doświadczeń na temat życia, kariery i problemów (niekoniecznie młodego) programisty.
Wesprzyj mnie
Jeżeli znalezione tutaj treści sprawiły, że masz ochotę wesprzeć moją działalność online, to zobacz na ile różnych sposobów możesz to zrobić. Niezależnie od tego co wybierzesz, będę Ci za to ogromnie wdzięczny.
Na wsparciu możesz także samemu zyskać. Wystarczy, że rzucisz okiem na listę różnych narzędzi, które używam i polecam. Decydując się na skorzystanie z któregokolwiek linku referencyjnego otrzymasz bonus również dla siebie.
Picture Credits
Dodaj komentarz