7,540
edits
Changes
no edit summary
==Mi az Avro?==
Az Avro egy nyílt forráskódú project, ami egy adata szerializációs szolgáltatás elsősorban az Apache Hadoop-hoz, de nem csak a Hadoop-ban használható, ahogy a mi példánkban is látni fogjuk. Av Az Avro segítségével nagyon hatékonyan cserélhetünk adatokat két végpont között "big data" környezetben.
Az Avro alapja egy séma regiszter, mihez amihez mind az adat szerializáló és deszerializáló szolgáltatás kapcsolódik. Itt tárolja az Avro a serializálnadó adatok tervrajzát JSON formátumban, a sémákat verziózva. Amikor a serializáló szolgáltatás adatot akar küldeni, akkor megjelöli az Avro serializátornak hogy melyik séma alapján serializálja a küldendő adatot. Ha a séma még nem létezett, akkor beszúrja a séma regiszterbe. Az Avro binárist készít a séma segítségével a küldendő adatból, és az üzenetbe a bináris adat mellé beleteszi a séma azonosítóját is, amit a deszerializiós szolgáltatás megkap, és annak segítségével ki tudja olvasni az adat deszerializálásához szükséges sémát, ami segítségével előállítja az eredeti üzenetet.
Az Avro séma regiszter több verziót is képes kezelni egy sémából. A beállításoknak megfelelően a séma lehet előre vagy visszafelé kompatibilis. Ha egy séma visszafelé kompatibilis, akkor az új sémával is ki lehet olvasni olyan régi adatokat, amit még egy korábbi sémával írtak be.
Az Avor sémákat JSON formátumban kell leírni, és aminek egyedi, kötött szintaktikájuk van, tehát nem szabványos JSON sémák. A schema-registry-vel el egy REST API-n keresztül lehet kommunikálni. A legtöbb programozási nyelven elérhető Avro magas szintű API, ami elfedi előlünk a REST kommunikációt. Az Avró többféle adatbázisban is tárolhatja a sémákat, de a leggyakoribb megoldás, hogy egy speciális Kafka topic-ban tárolja azokat.
<source lang="C++">
version: '3.2'
Hozzuk létre a docker stack-et:
<pre>
# docker stack deploy -c confluence_swarmconfluent_swarm.yaml confluenceconfluent
</pre>
# docker service ls
ID NAME MODE REPLICAS IMAGE PORTS
7vjvop7tqiyc confluence_kafka confluent_kafka replicated 1/1 confluentinc/cp-kafka:5.1.2 *:29092->29092/tcpin6a4ti3jeu5 confluence_schemaregistry confluent_schemaregistry replicated 1/1 confluentinc/cp-schema-registry:5.1.2 *:8081->8081/tcpoxxjtkcusj1f confluence_zookeeper confluent_zookeeper replicated 1/1 confluentinc/cp-zookeeper:5.1.2 *:32181->32181/tcp
</pre>
<pre>
NETWORK ID NAME DRIVER SCOPE
...
5albky0eu1to confluence_kafkaconfluent_kafka-net overlay swarm
olqkh5zlqiac ingress overlay swarm
...
=Avro REST interfész=
Az Avro a '''_schemas''' nevű Kafka topic-ban tárolja a sémákat az alapértelmezett konfiguráció szerint. Tehát az AVRO schema-registry-nek szüksége van . . A Kafka /bin mappájában található '''kafka-topics.sh''' topic admin script-el listázzuk ki a topikokat:
A válaszban láthatjuk a kompatibilitási szintet(ezt majd később részletesen tárgyaljuk).
Avro-ban minden sémát egy úgynevezett subject-ek alá kell regisztrálni. Egy subject alatt ugyan azon séma különböző verzióit tároljuk. Tehát két teljesen különböző sémát nem lehet ugyan azon a ugyanazon subject alá berakni. Tehát mikor hasonló sémákat regisztrálunk ugyan azon ugyanazon subject alá, akkor különböző verziók fognak létrejönni ugyan ahoz ahhoz a sémához. Azt hogy mekkora a megengedett eltérés mértéke, a schema-registry server konfigurációja határozza meg.
A sémákat a POST:/subjects/<subject-name>/versions REST interfészen kell beküldeni. A POST body-ban a {schema: "...séma definicíó..."} formátumban kell megadni a sémát, ahol a séma definíció egy escape-lt belső json.
<pre>
$ curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"schema" : "... schema def..."}' http://192.168.42.42:8081/subjects/<subject-name>/versions
Szúrjuk be az Avro-ba az alábbi '''Employee''' sémát. A namespace majd a schema-to-java kód generálásánál lesz érdekes, ez fogja meghatározni a java csomagotgenerált kódban. A type mező mutatja meg, hogy összetett objektumumot vagy objektumot, sima stringet, vagy tömböt ír le a séma. A '''record''' jelenti az összetett objektumot. Az '''Employee''' nevű objektum négy mezőből áll.
<source lang="C++">
{"namespace": "hu.alerant.kafka.avro.message",
Szúrjuk ezt be szintén a '''test1''' subject alá:
<pre>
$ curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"schema" : "{\"namespace\": \"hu.alerant.kafka.avro.message\",\"type\": \"record\", \"name\": \"Employee\", \"fields\": [ {\"name\": \"firstName\", \"type\": \"string\"}, {\"name\": \"lastName\", \"type\": \"string\"}, {\"name\": \"age\", \"type\": \"int\"}]}"}' http://192.168.42.42:8081/subjects/test1/versions
A generált osztályba az Avro belegenrálja a sémát is, ez az amit majd a Kafak topic-ba dobás előtt a producer fel fog küldeni a schema-register servernek: szervernek. Employee.java
<source lang="java">
package hu.alerant.kafka.avro.message;
==Java avro-kafak producer==
A hagyományos Kafka java producer-hez képest csak pár különbség van az a java producer inicializálásban. Egyrészt meg kell adni, hogy mind a kulcsot, mind az üzenetet Avro-val akarjuk serializálni, másrészt meg kell adni az Avro schema-registry URL-jét. A /etc/hosts fájlba felvettük a worker0 swarm node IP címével a '''schema-registry''' host nevet.
<source lang="java">
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroSerializer.class);
Első alkalommal, mikor a producer be akar dobni egy üzenetet a Kafka topic-ba, felküldi a sémát a már látott POST:http://192.168.42.42:8081/subjects/<subject-name>/versions REST hívással, amit az avro java objektumból nyer ki. Ha a séma egy futás alatt nem változik, akkor többször nem küldi fel a sémát a schema-registry-be. Az A producer az avro subject-et automatiksuan nevét automatikusan képezi a topoci nevéből. Tehát egy topoc-ba csak a kompatibilitási szabályoknak megfelelő sémáknak megfelelő üzeneteket lehet bedobaniberakni. Arra nincs mód, hogy bárhogyan is megadjuk, hogy az adott objektum melyik subject melyik verziójának kell hogy megfeleljen, ezt teljesen elfedi előlünk az API.
'''''Összefoglalva, egy adott Kafka topic-ba, amit kommunikációra használnunk (tehát nem a séma tárolására) csak Avro kompatibilis sémáknak megfelelő objektumokat lehet beküldeni. Nem azért mert a topic nem bírna el másik sémából gyártott bináris üzenetet, hanem azért, mert az Avro API a topic nevéből képzi a subject nevét, és egy subject-en belül csak kompatibilis sémákat lehet tárolni.'''''
</source>
A parancssori kafka-avro consumer segítségével fogjuk kiolvasni a java producer által küldött üzeneteket. Futtassuk le a java producer-t majd indítsuk el a parancssori consumer-t. Az avro consumer csak annyiban különbözök a sima parancssori consumer-től, hogy a séma regiszter címét is meg kell adni.
<pre>
./kafka-avro-console-consumer --topic test-topic --zookeeper 192.168.42.42:32181 --property schema.registry.url="http://schema-registry:8081"
Mikor Java-ból küldünk Avron-n keresztül Kafka üzeneteket, akkor a producer létre fog hozni a topic nevével prefixe-lt subjet-eket, egyet a Kafak kulcsnak és egyet a hozzá tartozó értéknek autómatikusanautomatikusan, az első üzenet váltás után. A fenit fenti példa futtatása után listázzuk ki az összes Avro-s subject-et:
<pre>
$ curl -X GET -H "Content-Type: application/vnd.schemaregistry.v1+json" http://192.168.42.42:8081/subjects/
</pre>
Láthatjuk, hogy létrehozott a '''test-topic ''' prefixel egy subject-et a valu-nak és a Kafka kulcsnak is.
</source>
A fenti módosítással példa futtatásakor már csak a value sémáját fogja elküldeni a Kafka-ba írás előtt a producer a séma regiszternek.
<pre>
2019-04-02 13:30:44 DEBUG RestService:118 - Sending POST with input {"schema":"{\"type\":\"record\",\"name\":\"Employee\",\"namespace\":\"hu.alerant.kafka.avro.message\",\"fields\":[{\"name\":\"firstName\",\"type\":\"string\"},{\"name\":\"lastName\",\"type\":\"string\"},{\"name\":\"age\",\"type\":\"int\"},{\"name\":\"phoneNumber\",\"type\":\"string\"}]}"} to http://schema-registry:8081/subjects/test-topic-value/versions
==Command line producers==
https://docs.confluent.io/3.0.0/quickstart.html<br>
A Confluence Confluent oldaláról letölthető Kafka csomagban található parancssori kafka-avro producer és consumer is. Töltsük le a Confulence Confulent csomagot innen: https://www.confluent.io/download/
A '''kafka-avro-console-producer''' program a /bin mappában található. 4 paramétert kell kötelezően kitöltenünk:
}
</source>
Ezt majd meg kell adjuk egy soros egysoros alakban a '''kafka-avro-console-producer''' parancsban.
{{note|A '''kafka-avro-console-producer''' parancsban a konkrét Avro üzenetet nem lehet megadni. Miután kiadtuk a parancsot, az input-on fogja várni, hogy bírjuk JSON formátumban a sémának megfelelő üzenetet. Minden egyes Enter leütésre megpróbálja elküldeni amit az stdIn-re beírtunk}}
==Command line consumer==
A command line producer-el megegyezően, szintén a Confluence Confluent oldaláról letölthető Kafka csomagban találhatjuk meg a command line kafka-avro consumer-t.
A '''kafka-avro-console-consumer''' program a /bin mappában található. Használata nagyon hasonlít a producer-re, 3 kötelező paramétere van: