Kafka Connect, pe scurt

  • 6 minute de citit
  • 5 exemple de cod în JSON și YAML
  • 1 imagine

Disponibil și în English

Pe această pagină

Salut, acest articol este despre Kafka Connect!

Introducere

Kafka Connect este un instrument pentru transferul de date în flux între Kafka și alte sisteme. Este distribuit și scalabil implicit, iar fiind un instrument standardizat, există deja foarte mulți connectori disponibili.

Connectorii conectează Kafka la un sistem sau invers. Există două tipuri de connectori

  • Source: Connectorii source preiau date dintr-un sistem existent, de exemplu: MariaDB, PostgreSQL, S3, Jira și altele, și transmit datele într-unul sau mai multe topicuri Kafka.
  • Sink: Connectorii sink preiau datele din topicuri și le introduc într-un sistem nou, de exemplu: MongoDB, Snowflake, S3.

Dacă vrei să transmiți în flux evenimentele de change data capture din bazele tale de date, Debezium oferă connectori care permit exact asta. CDC este un jurnal doar cu adăugare care identifică modificările din baze de date; folosind un flux CDC poți replica sau reconstrui o bază de date, iar în plus poți reacționa la evenimente procesându-le într-un sistem extern.

Kafka Connect poate fi rulat în mod standalone sau distribuit, ca un cluster de workeri.

Are o interfață RESTful pentru interacțiune:

  • configurarea connectorilor
  • pornirea, oprirea, suspendarea connectorilor
  • vizualizarea stării connectorilor
  • resetarea offset-urilor connectorilor

Permite, de asemenea, aplicarea mai multor transformări asupra unui mesaj.

Apache Kafka are o documentație excelentă despre Kafka Connect.

Iată o diagramă a unui sistem construit cu Kafka Connect, care replică date din PostgreSQL și MariaDB în ElasticSearch. ElasticSearch oferă multe instrumente pentru căutarea în date, rapidă și cu o precizie bună.

kafka-connect.png

Rest API

Pentru referință, am copiat toate operațiile din documentația REST API și le-am pus într-un tabel.

Method Path Description  
GET /connectors returnează o listă a connectorilor activi.  
POST /connectors creează un connector nou.  
GET /connectors/{name} obține informații despre un connector specific.  
DELETE /connectors/{name} șterge un connector.  
GET /connectors/{name}/config obține parametrii de configurare pentru un connector specific.  
PUT /connectors/{name}/config actualizează parametrii de configurare pentru un connector specific.  
PATCH /connectors/{name}/config modifică parțial parametrii de configurare pentru un connector specific.  
GET /connectors/{name}/status obține starea curentă a connectorului.  
GET /connectors/{name}/tasks obține lista task-urilor care rulează în prezent pentru un connector.  
GET /connectors/{name}/tasks/{taskid}/status obține starea curentă a task-ului.  
PUT /connectors/{name}/pause suspendă connectorul și task-urile sale, ceea ce oprește procesarea mesajelor până la reluarea connectorului.  
PUT /connectors/{name}/stop oprește connectorul și închide task-urile sale.  
PUT /connectors/{name}/resume reia un connector suspendat sau oprit.  
POST /connectors/{name}/restart repornește un connector și instanțele task-urilor sale.  
POST /connectors/{name}/tasks/{taskId}/restart repornește un task individual.  
PUT /connectors/{name}/topics/reset trimite o cerere de golire a setului de topicuri active ale unui connector.  
GET /connectors/{name}/offsets obține offset-urile curente pentru un connector.  
DELETE /connectors/{name}/offsets resetează offset-urile pentru un connector.  
GET /connector-plugins returnează o listă a pluginurilor de connector instalate în clusterul Kafka Connect.  
GET /connector-plugins/{plugin-type}/config obține definiția de configurare pentru pluginul specificat.  
PUT /connector-plugins/{connector-type}/config/validate validează valorile de configurare furnizate în raport cu definiția de configurare.  

Ca să pornești o instanță nouă de connector, de obicei folosești POST pe /connectors cu un corp de configurare:

{  "name": "my-jdbc-source",  "config": {    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",    "tasks.max": "1",    "connection.url": "jdbc:postgresql://localhost:5432/mydb",    "connection.user": "myuser",    "connection.password": "mypassword",    "mode": "incrementing",    "incrementing.column.name": "id",    "table.whitelist": "users",    "poll.interval.ms": "5000",    "topic.prefix": "pg.",    "value.converter": "org.apache.kafka.connect.json.JsonConverter",    "value.converter.schemas.enable": "false",    "transforms": "maskSensitive",    "transforms.maskSensitive.type": "org.apache.kafka.connect.transforms.MaskField$Value",    "transforms.maskSensitive.fields": "email,phone",    "transforms.maskSensitive.replacement": "****"  }}

Converters

Converterii sunt folosiți de connect pentru a converti valori dintr-un tip în altul. Converterii se aplică cheii și valorii mesajului Kafka. De exemplu, dacă ai următorul mesaj JSON:

{"data": 1}

Un string converter va pune mesajul în topicul Kafka ca string, în timp ce un JSON converter îl va păstra ca JSON. Există și converteri în format binar, precum Avro și ProtoBuf, care ajută la reducerea dimensiunii mesajului, împachetându-l într-un format compact. Un dezavantaj al acestui format este că ai nevoie de schema mesajului ca să-l deserializezi.

Poți, de asemenea, să scrii propriul converter și să-l încarci în Kafka Connect.

Pentru a seta converterii folosești următoarele chei:

  • key.converter: Setează converterul pentru cheia mesajului.
  • value.converter: Setează converterul pentru valoarea mesajului.

Iată câteva clase de converter uzuale:

  • org.apache.kafka.connect.storage.StringConverter
  • org.apache.kafka.connect.json.JsonConverter
  • org.apache.kafka.connect.converters.ByteArrayConverter
  • io.confluent.connect.json.JsonSchemaConverter (necesită schema registry)
  • io.confluent.connect.protobuf.ProtobufConverter (necesită schema registry)
  • io.confluent.connect.avro.AvroConverter (necesită schema registry)

De obicei setezi un converter astfel:

{  "key.converter": "org.apache.kafka.connect.json.JsonConverter",  "value.converter": "org.apache.kafka.connect.json.JsonConverter",  "value.converter.schemas.enable": "true",  "key.converter.schemas.enable": "true"}

Setând și value.converter.schemas.enable la true, vei primi schema mesajului JSON împreună cu payload-ul.

Schema Registry

Schema registry este o altă componentă, care funcționează ca un cache pentru schemele mesajelor.

Formatele binare precum Avro sau Protobuf nu pot fi decodate de receptor fără schema mesajului, iar trimiterea schemei cu fiecare mesaj mărește dimensiunea mesajului.

Scopul schema registry este să păstreze toate schemele împreună, într-o bază de date, și să permită producătorilor și consumatorilor să ceară schema doar când este nevoie, astfel încât mesajele să poată fi produse în topicul Kafka fără să includă schema.

Această componentă este opțională și este necesară doar când folosești formate binare precum Avro sau Protobuf.

Transforms

Poți aplica diverse transformări asupra mesajelor procesate de connector.

Transformările uzuale includ mascarea câmpurilor, eliminarea câmpurilor, înlocuirea valorilor, redenumirea câmpurilor și altele.

  • Cast - Convertește câmpuri sau întreaga cheie sau valoare la un tip specific
  • DropHeaders - Elimină header-e după nume
  • ExtractField - Extrage un câmp specific dintr-un Struct sau Map și include în rezultate doar acest câmp
  • Filter - Elimină mesajele din orice procesare ulterioară. Se folosește cu un predicat pentru a filtra selectiv anumite mesaje
  • Flatten - Aplatizează o structură de date imbricată
  • HeaderFrom - Copiază sau mută câmpuri din cheie sau valoare în header-ele înregistrării
  • HoistField - Învelește întregul eveniment ca un singur câmp într-un Struct sau Map
  • InsertField - Adaugă un câmp folosind fie date statice, fie metadate ale înregistrării
  • InsertHeader - Adaugă un header folosind date statice
  • MaskField - Înlocuiește câmpul cu o valoare null validă pentru tip (0, șir gol etc.) sau cu o înlocuire personalizată (doar șir nevid sau valoare numerică)
  • RegexRouter - Modifică topicul unei înregistrări pe baza topicului original, a unui șir de înlocuire și a unei expresii regulate
  • ReplaceField - Filtrează sau redenumește câmpuri
  • SetSchemaMetadata - Modifică numele sau versiunea schemei
  • TimestampConverter - Convertește timestamp-urile între formate diferite
  • TimestampRouter - Modifică topicul unei înregistrări pe baza topicului original și a timestamp-ului. Util când un sink trebuie să scrie în tabele sau indexuri diferite în funcție de timestamp-uri
  • ValueToKey - Înlocuiește cheia înregistrării cu o cheie nouă formată dintr-un subset de câmpuri din valoarea înregistrării

Sursa: https://kafka.apache.org/41/kafka-connect/user-guide/#transformations

Pentru a aplica transformări, le incluzi în configurarea connectorului:

{  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",  "transforms": "maskSensitive",  "transforms.maskSensitive.type": "org.apache.kafka.connect.transforms.MaskField$Value",  "transforms.maskSensitive.fields": "sensitiveField",  "transforms.maskSensitive.replacement": "****"}

Docker Compose

Poți porni o instanță Kafka Connect preconfigurată, împreună cu un cluster Kafka, pentru dezvoltare sau experimentare, folosind acest fișier docker-compose.

services:  broker:    image: confluentinc/cp-kafka:8.0.0    hostname: broker    container_name: broker    ports:      - "9092:9092"      - "9101:9101"    environment:      KAFKA_NODE_ID: 1      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092'      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1      KAFKA_JMX_PORT: 9101      KAFKA_JMX_HOSTNAME: localhost      KAFKA_PROCESS_ROLES: 'broker,controller'      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'      KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092'      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'      # Înlocuiește CLUSTER_ID cu un UUID base64 unic, generat cu "bin/kafka-storage.sh random-uuid"      # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'#  schema-registry:#    image: confluentinc/cp-schema-registry:8.0.0#    hostname: schema-registry#    container_name: schema-registry#    depends_on:#      - broker#    ports:#      - "8081:8081"#    environment:#      SCHEMA_REGISTRY_HOST_NAME: schema-registry#      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'#      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081  connect:    image: confluentinc/cp-kafka-connect:8.1.2    hostname: connect    container_name: connect    depends_on:      - broker#      - schema-registry    ports:      - "8083:8083"    environment:      CONNECT_BOOTSTRAP_SERVERS: 'broker:29092'      CONNECT_REST_ADVERTISED_HOST_NAME: connect      CONNECT_GROUP_ID: compose-connect-group      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter#      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"

Alternativ, poți descărca arhiva binară Kafka și rula connect cu bin/connect-standalone.sh config/connect-standalone.properties.

Atât. Sper că acest articol ți-a oferit o imagine de ansamblu asupra Kafka Connect și a capacităților sale.

References