Bird's rzeczywiste webhooki zdarzeń są niezwykle wartościowym narzędziem dla nadawców, pozwalającym na automatyczne przesyłanie danych do ich systemów. To może napędzać dalszą automatyzację, taką jak aktualizacja list mailingowych, wyzwalanie zautomatyzowanych podróży e-mailowych lub zasilanie wewnętrznych kokpitów. Chociaż te same dane zdarzenia można uzyskać za pośrednictwem interfejsu użytkownika Bird, korzystając z Event Search, lub programowo, korzystając z Events API Bird, ograniczenia dotyczące liczby rekordów zwracanych w jednym żądaniu lub limity szybkości nałożone na punkt końcowy API mogą sprawić, że obie te metody będą restrykcyjne dla dużych i zaawansowanych nadawców.
Webhooki zdarzeń w czasie rzeczywistym umożliwiają nadawcy skonfigurowanie punktu końcowego, do którego Bird przekaże dane i które można konsumować bez potrzeby planowania zadań cron, które pobierają dane. Istnieją również logistyczne kompromisy, gdy pobiera się dane w przeciwieństwie do ich dostarczania, takie jak konieczność identyfikacji, jaki okres czasu i parametry wykorzystać do każdego żądania API. Jeśli okresy czasu nie są idealnie zestrojone, ryzykujesz utratą danych, a jeśli się nakładają, musisz obsłużyć zduplikowane rekordy danych. Z real-time webhookami, dane zdarzeń są po prostu przesyłane do Twojego punktu końcowego, gdy tylko stają się dostępne w tej usłudze Bird.
Chociaż korzyści płynące z otrzymywania danych zdarzeń w czasie rzeczywistym do napędzania dalszych procesów automatyzacji mogą być natychmiast zrozumiane przez wielu nadawców, rzeczywisty proces wdrażania i konsumpcji webhooków może być zastraszający. Może to być szczególnie prawdziwe, jeśli nie znasz się na technicznych komponentach tworzenia punktu końcowego i programowego manipulowania danymi. Istnieją dostępne usługi, które konsumują dane webhooków Bird i automatycznie wprowadzają je do bazy danych używając mechanizmu ETL – przykładem może być usługa StitchData, o której pisaliśmy na blogu w przeszłości. Jeżeli jednak chcesz mieć większą kontrolę nad tym procesem, możesz łatwo samodzielnie zbudować te komponenty. Poniższy przewodnik ma na celu pomoc nadawcom, aby poczuli się komfortowo przy tworzeniu webhooka zdarzeń Bird i wykorzystywaniu danych przy użyciu infrastruktury wewnątrz AWS.
Konfiguracja docelowego punktu końcowego Webhooka
Gdy zdarzenie Bird zostaje utworzone, chcemy, aby strumień danych zdarzeń w czasie rzeczywistym trafił do punktu końcowego w AWS, abyśmy mogli programowo konsumować i wykorzystywać te dane. Dane zostaną wysłane z Bird do docelowego punktu końcowego, który przekaże ładunek do funkcji lambda, która przetworzy i przechowa dane w koszyku S3. Poniżej można zobaczyć diagram na wysokim poziomie z opisanym przepływem danych:
Aby wdrożyć ten przepływ pracy, zbudujmy je w odwrotnej kolejności, zaczynając od utworzenia koszyka S3, w którym będziemy przechowywać nasze dane zdarzeń, a potem pracujmy wstecz – dodając każdy komponent zasilający to, co już zbudowaliśmy.
Utwórz koszyk S3 do przechowywania danych Webhooka
Przed utworzeniem naszego load balancera do akceptacji danych, czy naszej funkcji lambda do przechowywania danych, musimy najpierw utworzyć nasz koszyk S3, w którym przechowywane będą dane. Aby to zrobić, przejdź do usługi S3 w AWS i naciśnij „Create Bucket.” Zostaniesz poproszony o nadanie nazwy koszykowi i ustawienie regionu – upewnij się, że używasz tego samego regionu, co Twój ALB i funkcja lambda. Kiedy Twój koszyk S3 zostanie utworzony, będzie pusty – jeśli chcesz zorganizować dane wewnątrz folderu, możesz teraz utworzyć zamierzoną ścieżkę lub katalog zostanie utworzony, gdy Twoja funkcja lambda przechowa plik. W tym przykładzie nazwaliśmy nasz koszyk S3 „bird-webhooks” i utworzyliśmy katalog o nazwie „B Event Data”, aby przechowywać nasze dane zdarzeń – zobaczysz te nazwy wspomniane poniżej w naszej funkcji lambda.
Utwórz funkcję Lambda do konsumpcji danych
Rzeczywiste przetwarzanie i przechowywanie danych będzie realizowane przez funkcję lambda, która jest wywoływana przez nasz load balancer aplikacji (ALB).
Pierwszym krokiem jest utworzenie funkcji lambda przez nawigację do usługi Lambda w AWS i kliknięcie „Create Function.” Zostaniesz poproszony o nadanie nazwy swojej funkcji lambda oraz wybranie, w jakim języku programowania napisać swoją funkcję. W tym przykładzie używamy Python jako języka wykonywalnego.
Teraz musimy rozwinąć naszą funkcję lambda. Przez chwilę załóżmy, że nasz load balancer aplikacji został skonfigurowany i przekazuje ładunek webhooka do naszej funkcji lambda – lambda otrzyma ładunek zawierający pełne nagłówki i ciało. Ładunek jest przekazywany do naszej funkcji lambda, używając obiektu „event” jako słownika. Możesz odwołać się do nagłówków i ciała ładunku niezależnie, uzyskując dostęp do obiektów „headers” i „body” w ramach ładunku. W tym przykładzie po prostu zamierzamy odczytać nagłówek „x-messagesystems-batch-id”, gdzie batch ID jest unikalną wartością utworzoną przez Bird dla partii webhooka, i użyć go jako nazwy pliku przy przechowywaniu ciała jako płaskiego pliku w S3; jednak możesz chcieć dodać dodatkowe funkcjonalności, takie jak kontrole autoryzacji czy obsługa błędów, w razie potrzeby.
Podczas przechowywania ładunku w postaci płaskiego pliku na S3, musimy zdefiniować nazwę koszyka S3, lokalizację i nazwę pliku, w którym przechowywane będą dane z ładunku. W naszym przykładowym funkcji lambda robimy to w ramach „store_batch” funkcji. W tym przykładzie zamierzamy przechowywać całą partię jako jeden plik, co pomaga zapewnić, że dane są zebrane i przechowane przed wygaśnięciem połączenia HTTP między Bird a twoim punktem końcowym. Chociaż możesz dostosować ustawienia limitu czasu połączenia na swoim load balancerze, nie ma gwarancji, że połączenie nie wygaśnie po stronie transmisji (w tym przypadku Bird) lub że połączenie nie zostanie zakończone przed zakończeniem wykonywania twojej funkcji lambda. Najlepszą praktyką jest, aby funkcja konsumenta była jak najbardziej wydajna i zachować działania przetwarzania danych dla procesów downstream, gdzie to możliwe – takich jak konwersja wsadowo sformatowanego JSON na plik CSV, lub załadowanie danych o zdarzeniach do bazy danych.
Ważne jest, aby pamiętać, że możliwe jest potrzebowanie aktualizacji zezwoleń dla swojej funkcji lambda. Twoja rola wykonawcza będzie potrzebować uprawnień PutObject i GetObject dla S3. Najlepszą praktyką jest egzekwowanie zasady najmniejszych uprawnień, więc zaleca się ustawić te uprawnienia tylko dla koszyka S3, w którym będą przechowywane ładunki webhooków.
Przykład naszej funkcji lambda konsumenta można znaleźć tutaj.
Jako krótka notatka na temat batch ID: Bird będzie ładować zdarzenia do jednego ładunku, gdzie każda partia może zawierać od 1 do 350 lub więcej rekordów zdarzeń. Batch zostanie przypisany unikalny batch ID, który można użyć do sprawdzenia statusu batch poprzez wykorzystanie Event Webhooks API lub wewnątrz swojego konta Bird, klikając na webhook stream i wybierając „Batch Status.” W przypadku, gdy ładunek webhooka nie mógł zostać dostarczony, na przykład podczas wygaśnięcia połączenia, Bird automatycznie ponownie wyśle batch z tym samym batch ID. Może się to zdarzyć, gdy twoja funkcja lambda działa blisko maksymalnego czasu rundy-odpływu wynoszącego 10 sekund i jest to powód do optymalizacji funkcji konsumenta w celu skrócenia czasu wykonywania.
Aby obsłużyć wszystkie działania przetwarzania danych, zaleca się utworzenie oddzielnej funkcji lambda, która będzie się wykonywać za każdym razem, gdy nowy plik zostanie utworzony w koszyku S3 – w ten sposób, przetwarzanie danych jest wykonywane asynchronicznie w stosunku do transmisji danych i nie ma ryzyka utraty danych z powodu zakończenia połączenia. Funkcję przetwarzania lambda opisuję w późniejszej sekcji.
Utwórz Load Balancer aplikacji
Aby otrzymać ładunek webhooka, musimy zapewnić punkt końcowy do wysyłania ładunków. Robimy to poprzez utworzenie balancera obciążenia aplikacji w AWS przez nawigację do EC2 > Load Balancers i kliknięcie „Create Load Balancer.” Będziesz poproszony o wybór, jaki rodzaj load balancera chcesz utworzyć – do tego chcemy utworzyć load balancer aplikacji. Musimy użyć balancera obciążenia aplikacji (ALB), aby zbudować nasz konsument, ponieważ webhooki zdarzeń będą wysyłane jako żądanie HTTP, a ALB są używane do kierowania żądań HTTP w AWS. Można by wdrożyć bramę HTTP jako alternatywę; jednak używamy ALB do tego projektu, ponieważ jest to bardziej lekkie i bardziej opłacalne niż brama HTTP. Ważne jest, aby zauważyć, że jeśli wybierzesz użycie bramy HTTP, format zdarzeń może być inny niż z ALB, a zatem twoja funkcja lambda będzie musiała obsłużyć odpowiednio obiekt żądania.
Gdy twój ALB zostanie utworzony, będziesz poproszony o nadanie nazwy swojemu ALB oraz skonfigurowanie schematu i ustawień dostępu/bezpieczeństwa – ponieważ planujemy otrzymywać dane zdarzeń z zewnętrznego źródła (Bird), chcemy, aby nasz ALB był publicznie dostępny w Internecie. W zakładce „Listeners and routing” ALB powinien nasłuchiwać HTTPS na porcie 443, i chcemy utworzyć grupę docelową, która wskazuje na naszą funkcję lambda, aby nasz ALB przekazywał przychodzące żądania do funkcji konsumenta lambda, którą stworzyliśmy powyżej. Musisz również upewnić się, że grupa bezpieczeństwa ma upoważnienie do zaakceptowania ruchu przez port 443.
Utwórz rekord DNS dla Load Balancer
Aby ułatwić nam użycie naszego ALB jako punktu końcowego, utworzymy rekord A w DNS, który wskaże na nasz ALB. Do tego możemy użyć usługi AWS Route 53 (lub obecnego dostawcy DNS) i utworzyć rekord A dla nazwy hosta, którą chcesz używać dla swojego punktu końcowego (np. spevents.<your_domain>). Rekord A powinien być skonfigurowany do wskazywania na ALB, który stworzyliśmy. Jeśli używasz Route 53 do zarządzania rekordami DNS, możesz odwołać się bezpośrednio do instancji ALB, włączając „Alias” i wybierając ALB; w przeciwnym razie, jeśli używasz zewnętrznego dostawcy DNS, powinieneś skierować rekord A na publiczny adres IP instancji ALB.
Zalecam użycie narzędzia, takiego jak Postman, aby przetestować, czy wszystko zostało poprawnie skonfigurowane przed włączeniem swojego webhooka Bird. Możesz wykonać żądanie POST do swojego punktu końcowego i potwierdzić, że otrzymano odpowiedź. Jeśli twoje żądanie POST nie zwraca odpowiedzi, można potrzebować dwukrotnie sprawdzić, czy twój ALB nasłuchuje na właściwym porcie.
Utwórz webhook Bird
Teraz jesteśmy gotowi, aby utworzyć webhook w Bird i użyć nazwy hosta zdefiniowanej w rekordzie A powyżej jako naszego docelowego punktu końcowego. Aby utworzyć webhook, przejdź do sekcji Webhooks w ramach swojego konta Bird i kliknij „Create Webhook.” Zostaniesz poproszony o przypisanie nazwy swojemu webhookowi i podanie docelowego URL – celem powinien być nazwa hosta rekordu A, który utworzyłeś wcześniej. Należy zauważyć, że docelowy URL może wymagać, aby przedrostek „HTTPS://” był zawarty w URL.
Po zakończeniu, zweryfikuj, że wybrano odpowiednie subkonto i zdarzenia, i naciśnij „Create Webhook” aby zapisać konfigurację. Teraz dane zdarzenia dla wszystkich wybranych typów zdarzeń będą przesyłane do naszego docelowego URL i konsumowane przez nasz ALB dla dalszego przetwarzania.
Przetwarzanie danych zdarzeń Webhooka
W zależności od zamierzonego celu przechowywania danych zdarzeń Bird, twoje wymagania mogą być spełnione przez samą prostą operację przechowywania ładunku JSON jako płaskiego pliku. Możesz również mieć już ustanowiony proces ETL downstream, zdolny do konsumpcji i ładowania danych w formacie JSON. W obu tych przypadkach, możesz być w stanie wykorzystać powstały plik płaski, utworzony przez naszą funkcję przetwarzania lambda, bez zmian.
Alternatywnie, możesz potrzebować transformować dane – na przykład przekształcić z format JSON na format CSV – lub ładować dane bezpośrednio do bazy danych. W tym przykładzie stworzymy prostą funkcję lambda, która przekształca dane webhooka z oryginalnego formatu JSON na plik CSV, który może być załadowany do bazy danych.
Utwórz funkcję Lambda do przetwarzania danych
Podobnie jak przy funkcji lambda do konsumpcji danych z webhooka, musimy utworzyć nową funkcję lambda poprzez przejście do usługi Lambda w AWS i kliknięcie „Create Function.” Nowa funkcja lambda zostanie wyzwolona, gdy nowy plik zostanie utworzony w naszym koszyku S3 – odczyta ona dane i przekształci je w nowy plik csv.
Funkcja lambda przyjmuje informacje o pliku jako zdarzenie. W przykładowej funkcji lambda zobaczysz, że najpierw mamy serię kontroli walidacji, aby upewnić się, że dane są kompletne i uformowane zgodnie z oczekiwaniami. Następnie przekształcamy ładunek JSON na plik CSV, korzystając z biblioteki „csv” i zapisując do pliku tymczasowego. Funkcje lambda mogą zapisywać lokalne pliki tylko w katalogu „/tmp”, dlatego tworzymy tymczasowy plik csv i nazywamy go za pomocą konwencji <batch_id>.csv. Powodem, dla którego używamy batch_id jest po prostu zapewnienie, że wszystkie procesy równoległe uruchamiane w wyniku odbierania wielu ładunków webhooka nie będą sobie przeszkadzać, ponieważ każda partia webhooka będzie miała unikalny batch_id.
Po pełnym przekształceniu danych na CSV, odczytujemy dane CSV jako strumień bajtowy, usuwamy plik tymczasowy i zapisujemy dane CSV jako nowy plik na S3. Ważne jest, aby pamiętać, że do przechowywania wynikowych danych potrzebny jest inny koszyk S3, w przeciwnym razie ryzykujemy stworzeniem pętli rekurencyjnej, która może powodować zwiększone użycie funkcji lambda i zwiększone koszty. Musimy zidentyfikować, w którym koszyku S3 i lokalizacji chcemy przechowywać nasz plik CSV w ramach naszej funkcji lambda. Postępuj według tej samej procedury co powyżej, aby stworzyć nowy koszyk S3 do przechowywania naszego pliku CSV.
Pamiętaj, że katalog tmp ma ograniczoną przestrzeń do 512 MB, dlatego ważne jest, aby plik tymczasowy był usunięty po użyciu, aby zapewnić wystarczającą ilość miejsca na przyszłe wykonania. Powodem, dla którego używamy pliku tymczasowego zamiast bezpośredniego zapisywania na S3 jest uproszczenie połączenia z S3 poprzez posiadanie jednego żądania.
Podobnie jak w przypadku funkcji lambda konsumenta, możesz potrzebować zaktualizować uprawnienia dla swojej funkcji lambda przetwarzającej. Ta funkcja lambda wymaga, aby rola wykonawcza miała uprawnienia GetObject dla wejściowego koszyka S3 oraz uprawnienia PutObject i GetObject dla wyjściowego koszyka S3.
Przykład naszej funkcji lambda przetwarzającej można znaleźć tutaj.
Skonfiguruj funkcję Lambda, aby wykonywała się, gdy nowe dane są przechowywane na S3
Teraz, gdy nasza funkcja lambda do konwersji pliku z formatu JSON na CSV została utworzona, musimy skonfigurować ją, aby była wyzwalana, gdy nowy plik zostanie utworzony w naszym koszyku S3. Aby to zrobić, musimy dodać wyzwalacz do naszej funkcji lambda, otwierając funkcję lambda i klikając „Add Trigger” na górze strony. Wybierz „S3” i podaj nazwę koszyka S3, w którym przechowywane są surowe ładunki webhooków. Masz także możliwość określenia prefiksu i/lub sufiksu pliku do filtrowania. Po skonfigurowaniu ustawień możesz dodać wyzwalacz, klikając „Add” na dole strony. Teraz twoja funkcja lambda przetwarzająca będzie się wykonywać, gdy nowy plik zostanie dodany do twojego koszyka S3.
Ładowanie danych do bazy danych
W tym przykładzie nie omówię szczegółowo ładowania danych do bazy danych, ale jeśli podążasz za tym przykładem, masz kilka opcji:
Załaduj dane bezpośrednio do swojej bazy danych w ramach swojej funkcji lambda przetwarzającej
Skommutujesz swój plik CSV używając ustanowionego procesu ETL
Niezależnie od tego, czy korzystasz z usługi bazy danych AWS, takiej jak RDS lub DynamoDB, czy masz swoją własną bazę danych PostgreSQL (lub podobną), możesz połączyć się z swoją usługą bazy danych bezpośrednio z funkcji lambda przetwarzającej. Na przykład w ten sam sposób, w jaki wywołaliśmy usługę S3, używając „boto3” w naszej funkcji lambda, możesz również użyć „boto3” do wywołania RDS lub DynamoDB. Usługa AWS Athena może również być używana do odczytu plików danych bezpośrednio z plików płaskich i dostępu do danych za pomocą języka zapytań podobnego do SQL. Zalecam odniesienie się do odpowiedniej dokumentacji dla usługi, której używasz, aby uzyskać więcej informacji o najlepszym sposobie realizacji tego w twoim środowisku.
Podobnie istnieje wiele usług, które mogą pomóc w skommutowaniu plików CSV i ładowaniu danych do bazy danych. Możesz już mieć ustanowiony proces ETL, który możesz wykorzystać.
Mamy nadzieję, że znalazłeś ten przewodnik pomocnym – miłego wysyłania!
