Conectorul sursă GitHub pentru Kafka Connect

Scris de Denis Nuțiu July 29, 2020

4 minute de citit
11 exemple de cod

Disponibil și în English

Pe această pagină

Salut!

În acest articol vom discuta cum să începi rapid cu Kafka și Kafka Connect, ca să colectezi toate commit-urile dintr-un repository GitHub.

Este un tutorial practic, care te scutește de timpul pierdut parcurgând documentația Kafka.

Mediul de lucru

Kafka este cam greu de configurat; vei avea nevoie cel puțin de Kafka, Zookeper și Kafka Connect. Ca să simplificăm configurarea, vom folosi Docker și docker-compose.

Odată ce ai Docker și docker-compose instalate pe sistem, poți rula o singură comandă și obții un mediu Kafka pentru dezvoltare.

Să pornim Kafka folosind repository-ul Confluent Platform. Asigură-te că ai Docker, Docker-Compose și Git instalate înainte de a continua.

git clone https://github.com/confluentinc/cp-all-in-onecd .\cp-all-in-one\cp-all-in-one-communitydocker-compose up

După ce comanda docker-compose up se termină, Kafka și componentele asociate ar trebui să ruleze. Ca să verifici că toate componentele funcționează corect, deschide o fereastră nouă de terminal și compară rezultatul comenzii docker ps cu al meu:

CONTAINER ID        IMAGE                                         COMMAND                  CREATED             STATUS                            PORTS                                                                      NAMESfc95d4828397        confluentinc/ksqldb-examples:5.5.1            "bash -c 'echo Waiti…"   6 minutes ago       Up 6 minutes                 ksql-datagen28e67d645888        confluentinc/cp-ksqldb-cli:5.5.1              "/bin/sh"                6 minutes ago       Up 6 minutes                 ksqldb-cli371fd742ad1f        confluentinc/cp-ksqldb-server:5.5.1           "/etc/confluent/dock…"   6 minutes ago       Up 6 minutes (healthy)            0.0.0.0:8088->8088/tcp                 ksqldb-servercaeb86c71308        cnfldemos/kafka-connect-datagen:0.3.2-5.5.0   "/etc/confluent/dock…"   6 minutes ago       Up 6 minutes (health: starting)   0.0.0.0:8083->8083/tcp, 9092/tcp                 connectc4d90361c677        confluentinc/cp-kafka-rest:5.5.1              "/etc/confluent/dock…"   6 minutes ago       Up 6 minutes                      0.0.0.0:8082->8082/tcp                 rest-proxyc3752a0c1487        confluentinc/cp-schema-registry:5.5.1         "/etc/confluent/dock…"   6 minutes ago       Up 6 minutes                      0.0.0.0:8081->8081/tcp                 schema-registry14bfbef71822        confluentinc/cp-kafka:5.5.1                   "/etc/confluent/dock…"   6 minutes ago       Up 6 minutes                      0.0.0.0:9092->9092/tcp, 0.0.0.0:9101->9101/tcp, 0.0.0.0:29092->29092/tcp   brokerf2ddb8971efa        confluentinc/cp-zookeeper:5.5.1               "/etc/confluent/dock…"   6 minutes ago       Up 6 minutes                      2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp                 zookeeper

Dacă arată similar, ești pregătit să începi configurarea Kafka Connect.

Kafka Connect

Kafka Connect este un instrument excelent pentru construirea de pipeline-uri de date. Îl vom folosi ca să aducem date de pe GitHub în Kafka.

Am putea scrie un simplu producer în Python care să interogheze API-ul GitHub și să producă un record pentru Kafka folosind o bibliotecă client, dar Kafka Connect vine cu beneficii suplimentare. Este cam ca Docker containers pentru producători și consumatori; de fapt, în Kafka Connect, producătorii se numesc conectori source, iar consumatorii conectori sink.

Instalarea conectorului GitHub

Ca să instalezi conectorul GitHub, susținut de Confluent, trebuie să îl descarci de aici. (dă click pe download zip)

După ce descărcarea s-a terminat, dezarhivează fișierele.

Înainte de instalare, dacă accesezi: vei vedea conectorii instalați în prezent; momentan nu există niciun conector GitHub.

Ca să îl instalezi, trebuie să copiem directorul lui în containerul Kafka Connect și să repornim containerul:

docker cp ./confluentinc-kafka-connect-github-1.0.1/. connect:/usr/share/javadocker restart connect

Dacă accesezi din nou http://localhost:8083/connector-plugins, vom vedea că conectorul a fost instalat cu succes:

{"class":"io.confluent.connect.github.GithubSourceConnector","type":"source","version":"1.0.1"},

Configurarea conectorului

Conectorul nu are încă nicio configurație, deci nu rulează. Ca să configurezi un conector, trebuie să trimiți o cerere POST către endpoint-ul de configurare. Poți avea oricâte configurații dorești pentru conector, fiecare reprezentând un worker de sine stătător; important este că trebuie să folosești un nume diferit pentru fiecare.

Ca să vezi conectorii deja configurați, putem accesa: localhost:8083/connectors/. Cum era de așteptat, ar trebui să primim un JSON gol.

Ca să configurez conectorul GitHub, am luat configurația de exemplu din documentația lui și am adaptat-o în format JSON.

Am ales, de asemenea, să folosesc JsonConverter în loc de AvroConverter, pentru simplitate, și în loc să iau stargazers, iau commit-urile repository-ului.

Iată configurația:

{"name":"MyGithubConnector","config":{"connector.class":"io.confluent.connect.github.GithubSourceConnector","confluent.topic.bootstrap.servers":"broker:29092","tasks.max":"1","confluent.topic.replication.factor":"1","github.service.url":"https://api.github.com","github.repositories":"apache/kafka","github.tables":"```commits```","github.since":"2019-01-01","github.access.token":"YOUR_PERSONAL_ACCESS_TOKEN","topic.name.pattern":"github-avro-${entityName}","key.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter":" org.apache.kafka.connect.json.JsonConverter"}}

Acum trebuie să încărcăm configurația în Kafka Connect. Am salvat-o într-un fișier numit github-connector-config.json și am folosit curl ca să o încarcăm. Dacă încărcarea reușește, serverul ar trebui să îți returneze configurația:

➜ cp-all-in-one-community git:(5.5.1-post) ✗ curl -X POST -H "Content-Type: application/json" --data @github-connector-config.json http://localhost:8083/connectors{"name":"MyGithubConnector","config":{"connector.class":"io.confluent.connect.github.GithubSourceConnector","confluent.topic.bootstrap.servers":"broker:29092","tasks.max":"1","confluent.topic.replication.factor":"1","github.service.url":"https://api.github.com","github.repositories":"apache/kafka","github.tables":"commits","github.since":"2019-01-01","github.access.token":"xxx","topic.name.pattern":"github-avro-${entityName}","key.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter":" org.apache.kafka.connect.json.JsonConverter","name":"MyGithubConnector"},"tasks":[],"type":"source"}%

Accesând acum http://localhost:8083/connectors/ se returnează configurația conectorului nostru GitHub:

["MyGithubConnector"]

Dacă verificăm starea lui (http://localhost:8083/connectors/MyGithubConnector/status), ar trebui să vedem că deja rulează:

{"name":"MyGithubConnector","connector":{"state":"RUNNING","worker_id":"connect:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"connect:8083"}],"type":"source"}

Conectorul va prelua toate commit-urile la care are acces; în cele 5 minute cât l-am lăsat să ruleze, a preluat peste 2,5 mii de commit-uri de pe GitHub.

Îl opresc ca să economisesc spațiu pe disc:

curl -X PUT localhost:8083/connectors/MyGithubConnector/pause

Inspectarea datelor din Kafka

Pentru a inspecta datele folosesc un instrument numit kafkacat.

Rularea comenzii kafkacat -C -b localhost:9092 -t github-avro-commits ar trebui să îți arate datele colectate de pe GitHub:

"commit":{ "author":{ "name":"hejiefang", "email":"xxxx", "date":1546437651000 }, "committer":{ "name":"Manikumar Reddy", "email":"xxxx", "date":1546437651000 }, "message":"MINOR: Fix doc format in upgrade notes\n\nAuthor: hejiefang \n\nReviewers: Srinivas , Manikumar Reddy \n\nCloses #6076 from hejiefang/modifynotable", "tree":{ "sha":"91db6646f829d6636011d09fdc8643e36280716b", "url":"https://api.github.com/repos/apache/kafka/git/trees/91db6646f829d6636011d09fdc8643e36280716b" }, "url":"https://api.github.com/repos/apache/kafka/git/commits/ffd6f2a2e8a573695d0c1c98e663f0b8198b1b6d", "comment_count":0, "verification":{ "verified":false, "reason":"unsigned", "signature":null, "payload":null } },

Rezumat

Am configurat cu succes un mediu simplu de joacă Kafka cu docker-compose și am instalat un conector SOURCE pentru GitHub. A fost destul de ușor, iar API-ul REST al Kafka Connect este foarte puternic și ușor de folosit.

În următorul articol vom lua datele despre commit-uri și le vom indexa în ElasticSearch folosind un conector SINK.

Mulțumesc că ai citit!

Bonus

Poți depana conectorul modificând nivelul de logging:

// Setare loggingcurl -s -X PUT -H "Content-Type:application/json" \http://localhost:8083/admin/loggers/io.confluent.connect.github.GithubSourceConnector \-d '{"level": "TRACE"}' \| jq '.'