{"id":32632,"date":"2019-10-31T21:48:08","date_gmt":"2019-10-31T18:48:08","guid":{"rendered":"https:\/\/prohoster.info\/blog\/ne-tolko-obrabotka-kak-my-sdelali-iz-kafka-streams-raspredelennuyu-bazu-dannyh-i-chto-iz-etogo-vyshlo\/"},"modified":"2019-10-31T21:48:08","modified_gmt":"2019-10-31T18:48:08","slug":"ne-tolko-obrabotka-kak-my-sdelali-iz-kafka-streams-raspredelennuyu-bazu-dannyh-i-chto-iz-etogo-vyshlo","status":"publish","type":"post","link":"https:\/\/prohoster.info\/nl\/blog\/administrirovanie\/ne-tolko-obrabotka-kak-my-sdelali-iz-kafka-streams-raspredelennuyu-bazu-dannyh-i-chto-iz-etogo-vyshlo","title":{"rendered":"Niet alleen verwerking: Hoe we van Kafka Streams een gedistribueerde database hebben gemaakt, en wat daaruit voortkwam","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

Hallo, Habr!<\/p>\n

We remind you that following the book on Kafka<\/a><\/noindex> we have released another fascinating work about the library Kafka Streams API<\/a><\/noindex>. <\/p>\n

\"Niet
\n
\nWhile the community is still exploring the boundaries of this powerful tool, a new article has been published that we want to introduce you to. The author shares from their own experience how to create a distributed data storage using Kafka Streams. Enjoy reading!
\n<\/a><\/noindex>
\nThe Apache library Kafka Streams<\/a><\/noindex> is used worldwide in enterprises for distributed stream processing over Apache Kafka. One of the underrated aspects of this framework is that it allows for the storage of local state generated based on stream processing.<\/p>\n

In this article, I will explain how our company successfully utilized this feature in developing a cloud application security product. Using Kafka Streams, we created microservices with shared state, each serving as a fault-tolerant and highly available source of reliable information about the state of objects in the system. This represents a step forward in both reliability and ease of maintenance for us.<\/p>\n

If you are interested in an alternative approach that allows for a single central database to maintain the formal state of your objects \u2013 read on, it will be interesting\u2026 <\/p>\n

Why we felt it was time to change our approaches to working with shared state <\/b><\/p>\n

We needed to maintain the state of various objects based on agent reports (for example: was the site under attack?). Before transitioning to Kafka Streams, we often relied on a single central database (+ service API) for state management. This approach has its drawbacks: in data-intensive situations<\/a><\/noindex> maintaining consistency and synchronization becomes a real challenge. The database can become a bottleneck or end up in a race condition<\/a><\/noindex> and suffer from unpredictability.<\/p>\n

\"Niet
\n
\nIllustration 1: a typical state separation scenario encountered before switching to
\nKafka and Kafka Streams: agents communicate their representations via the API, and the updated state is calculated through the central database <\/i><\/p>\n

Meet Kafka Streams \u2013 it has now become easy to create microservices with shared state. <\/b><\/p>\n

Ongeveer een jaar geleden besloten we onze scenario's voor het werken met gedeelde toestand grondig te herzien om enkele problemen aan te pakken. We besloten direct Kafka Streams uit te proberen \u2013 het is immers bekend hoe schaalbaar, hoog beschikbaar en fouttolerant het is, en hoe rijk de streamingfunctionaliteit is (inclusief transformaties met behoud van toestand). Het was precies wat we nodig hadden, om nog maar te zwijgen over hoe volwassen en betrouwbaar het berichtenuitwisselingssysteem van Kafka is.<\/p>\n

Elk van de door ons gemaakte microservices met behouden toestand werd gebouwd op basis van een Kafka Streams-instantie met een vrij eenvoudige topologie. Deze bestond uit 1) een bron 2) een processor met een permanente opslag voor sleutels en waarden 3) een sink:<\/p>\n

\"Niet
\n
\nIllustratie 2: de standaard topologie van onze streaminginstanties voor microservices met behouden toestand. Merk op: hier is ook een opslag aanwezig, waarin metadata over planning zich bevindt. <\/i><\/p>\n

Met deze nieuwe benadering stellen agents berichten op die naar de oorspronkelijke topic worden gestuurd, en consumenten \u2013 laten we zeggen, de service voor e-mailmeldingen \u2013 ontvangen de berekende gedeelde toestand via de sink (de uitgangstopic).<\/p>\n

\"Niet
\n
\nIllustratie 3: een nieuw voorbeeld van een takenstroom voor een scenario met gedeelde microservices: 1) een agent genereert een bericht dat binnenkomt in de oorspronkelijke Kafka-topic; 2) de microservice met gedeelde toestand (die Kafka Streams gebruikt) verwerkt het en schrijft de berekende toestand naar de eindtopic van Kafka; waarna 3) de consumenten de nieuwe toestand ontvangen.<\/i><\/p>\n

H\u00e9, en deze ingebouwde opslag voor sleutels en waarden is echt heel nuttig!<\/b><\/p>\n

Zoals hierboven vermeld, bevat onze topologie met gedeelde toestand een opslag voor sleutels en waarden. We hebben verschillende toepassingsmogelijkheden gevonden, waarvan er twee hieronder worden beschreven.<\/p>\n

Optie #1: gebruik van de opslag voor sleutels en waarden bij berekeningen<\/b><\/i><\/p>\n

Onze eerste key-value store bevatte de helperdata die we nodig hadden voor berekeningen. Bijvoorbeeld, in sommige gevallen werd de gedeelde toestand bepaald op basis van het 'meerderheidsstem'-principe. In de opslag konden we alle laatste rapporten van agenten over de toestand van een bepaald object bewaren. Vervolgens, wanneer we een nieuw rapport van een bepaalde agent kregen, konden we dit opslaan, rapporten van alle andere agenten over dezelfde toestand van dat object uit de opslag ophalen en de berekening herhalen.
\nHieronder in illustratie 4 wordt weergegeven hoe we toegang verleenden tot de key-value store voor de verwerkende methode van de processor, zodat we een nieuw bericht konden verwerken.<\/p>\n

\"Niet
\n
\nIllustratie 4: toegang verlenen tot de key-value store voor de verwerkende methode van de processor (daarna moet in elk scenario dat met gedeelde toestand werkt, de methode worden ge\u00efmplementeerd) doProcess<\/code>)<\/i><\/p>\n

Optie #2: een CRUD API bovenop Kafka Streams maken<\/b><\/i><\/p>\n

Nadat we onze basis taakstroom hadden opgezet, gingen we proberen een RESTful CRUD API te schrijven voor onze microservices met gedeelde toestand. We wilden de mogelijkheid hebben om de toestand van bepaalde of alle objecten op te halen, evenals de toestand van een object in te stellen of te verwijderen (dit is handig voor de ondersteuning van de serverzijde).<\/p>\n

Voor ondersteuning van alle API Get State, telkens wanneer we de toestand opnieuw moesten berekenen tijdens de verwerking, legden we deze lange tijd vast in de ingebouwde key-value store. In dit geval wordt het vrij eenvoudig om zo'n API te implementeren met een enkele instantie van Kafka Streams, zoals in de onderstaande listing wordt weergegeven:<\/p>\n

\"Niet
\n
\nIllustratie 5: gebruik van de ingebouwde key-value store voor het verkrijgen van de vooraf berekende toestand van een object. <\/i><\/p>\n

Het bijwerken van de toestand van een object via de API is ook niet moeilijk te implementeren. In principe is het voldoende om een Kafka producer te cre\u00ebren en hiermee een opname te maken waarin de nieuwe toestand is opgenomen. Dit garandeert dat alle berichten die via de API worden gegenereerd, op dezelfde manier worden verwerkt als die van andere producers (bijvoorbeeld agenten).<\/p>\n

\"Niet
\n
\nIllustratie 6: de toestand van een object kan worden ingesteld met behulp van een Kafka producer. <\/i><\/p>\n

Een kleine complicatie: Kafka heeft veel partitionen.<\/b><\/p>\n

Vervolgens wilden we de verwerking load verdelen en de beschikbaarheid verbeteren door voor elk scenario een cluster van microservices met een gedeelde status te bieden. De configuratie ging ons als vanzelf af: nadat we alle instanties zo hadden ingesteld dat ze met dezelfde app-ID werkten (en met dezelfde opstartservers), werd vrijwel alles automatisch afgehandeld. We hebben ook vastgesteld dat elke bron-topic uit meerdere partities zou bestaan, zodat elke instantie een subset van die partities kon toegewezen krijgen.<\/p>\n

Ik wil ook vermelden dat het gebruikelijk is om een back-up van de statusopslag te maken, zodat in het geval van een herstel na een storing deze kopie naar een andere instantie kan worden overgebracht. Voor elke statusopslag in Kafka Streams wordt een repliceerbare topic met een wijzigingslog gemaakt (waarbij lokale updates worden bijgehouden). Zo biedt Kafka voortdurend een back-up voor de statusopslag. Daarom kan de statusopslag van een bepaalde Kafka Streams-instantie snel worden hersteld op een andere instantie, waar de bijbehorende partities naartoe gaan. Onze tests hebben aangetoond dat dit in enkele seconden kan worden gedaan, zelfs als er miljoenen records in de opslag staan.<\/p>\n

Bij de overgang van \u00e9\u00e9n microservice met een gedeelde status naar een cluster van microservices wordt het iets minder triviaal om de Get State API te implementeren. In deze nieuwe situatie bevat de statusopslag van elke microservice slechts een deel van het totale beeld (de objecten waarvan de sleutels op een specifieke partitie waren geprojecteerd). We moesten bepalen op welke instantie de status van het object dat we nodig hadden zich bevond, en dit deden we op basis van de metadata van de streams, zoals hieronder weergegeven:<\/p>\n

\"Niet
\n
\nIllustratie 7: met behulp van de streammetadata bepalen we van welke instantie we de status van het benodigde object moeten opvragen; een dergelijke aanpak werd toegepast met de GET ALL API.<\/i><\/p>\n

Belangrijkste bevindingen<\/b><\/p>\n

Statusopslag in Kafka Streams kan de facto fungeren als een gedistribueerde database, <\/p>\n