Kas yra Kafka Connector?

Kafka Connector - tai Apache Kafka karkasas, skirtas integruoti Kafka su išorinėmis sistemomis (duomenų bazėmis, failų sistemomis, debesų saugyklomis ir kt.) be naudotojo kodo rašymo. Jis abstrakčiai apibūdina integracijos detales ir siūlo standartizuotą būdą duomenų perkėlimui.

Yra du konnektorių tipai:

• Source Connectors: įkelia duomenis iš išorinių šaltinių į Kafka.
• Sink Connectors: eksportuoja duomenis iš Kafka į išorines sistemas.

Pagrindinės Kafka Connect komponentės:

• Connect Workers: procesai, kuriuose veikia konnektoriai ir užduotys. Gali veikti kaip standalone (kuriant ir testuojant) arba kaip distributed (produkcinėje aplinkoje, su atsparumo ir mastelio galimybėmis) režimu.
• Connectors: loginis duomenų perkėlimo užduoties aprašymas. Nustato, kokius duomenis ir kur perkelti.
• Tasks: darbo vienetai, kurie faktiškai perkelia duomenis. Vienas konnektorius gali turėti kelis užduočių vienetus paraleliai apdorojimui.

Privalumai naudojant Kafka Connect:

• Supaprastina integraciją: nereikia rašyti daug boilerplate kodo.
• Patikimumas: palaiko atsparumą klaidoms, mastelį ir duomenų pristatymo garantijas (at-least-once arba exactly-once, priklausomai nuo konnektoriaus).
• Plėtojimas: lengva kurti savo konnektorius specifiniams poreikiams.
• Valdymas: konnektorius galima valdyti per REST API.

Pavyzdys naudojimo (hipotetinis, demonstracijai):

Tarkime, mums reikia perkelti duomenis iš PostgreSQL duomenų bazės į Kafka topiką. Galime naudoti Source Connectorį PostgreSQL:

# Pavyzdinis komandos pavyzdys konnektoriaus sukūrimui per REST API
# Tai supaprastintas pavyzdys, tikrieji parametrai gali skirtis
curl -X POST -H "Content-Type: application/json" --data '
{
  "name": "postgres-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://<db_host>:<db_port>/<db_name>",
    "connection.user": "<db_user>",
    "connection.password": "<db_password>",
    "topic.prefix": "postgres-data-",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "table.whitelist": "public.users"
  }
}' http://<connect_worker_host>:<connect_worker_port>/connectors
```}}}}}]}},{