Kafka Connect CDC: PostgreSQL ל-Elasticsearch ו-JDBC

הערה: כאשר מסד הנתונים PostgreSQL גדל למאות ג'יגה-בייט ודרישות הרעננות של אינדקס החיפוש יורדות לשניות, סנכרון ידני מפסיק לעבוד. אנו מגדירים את Kafka Connect לסנכרון נתונים בזרם (CDC) — זה אמין ומהיר יותר משירותי סקרים מותאמים אישית. לדוגמה, שוק עם קטלוג ב-PostgreSQL וחיפוש ב-Elasticsearch נתקל בעיכובי עדכון אינדקס של עד 15 דקות. עם הפתרון שלנו, העיכוב יורד ל-2 שניות (שיפור של 99.8%), וכאשר צומת נופל, נתונים לא אובדים — משימות מתחלקות מחדש אוטומטית. מקרה טיפוסי אחד: PostgreSQL → Debezium → Kafka → Elasticsearch. ללא Kafka Connect, מהנדסים מבזבזים שבועות בכתיבת handler ל-WAL והתמודדות עם כפילויות. אנו עושים זאת ב-4 ימים turnkey, עם קלאסטר עמיד לתקלות וניטור. תיעוד Debezium מאשר ש-CDC מספק סטרימינג עם חביון מינימלי. הפתרון שלנו חוסך עד $15,000 בשנה בעלויות תחזוקה בהשוואה לסקרים מותאמים אישית.

למה Kafka Connect עדיף על אינטגרציה מותאמת אישית?

נשווה בין הגישות:

תוצאה: Kafka Connect מספק מסגרת מוכנה עם אספקה מובטחת (exactly-once עם יצרנים idempotent) וחוסך 80% מזמן הפיתוח ו-50% מזמן התפעול.

איך CDC עובד על PostgreSQL עם Debezium?

Change Data Capture (CDC) מיירט כל שינוי במסד הנתונים ומשדר אותו כאירועים. Debezium מתחבר לPostgreSQL WAL (Write-Ahead Log) ושולח INSERT/UPDATE/DELETE ל-Kafka. זה מבטל את הצורך בטריגרים מותאמים אישית או בסקרים על טבלאות בלוח זמנים. Debezium תומך ב-snapshot.mode=initial לטעינה ראשונית וב-incremental כדי למנוע נעילות על טבלאות גדולות. לאחר ההגדרה, כל שינוי מופיע בנושא Kafka תוך אלפיות שניות. החביון הממוצע הוא 100 אלפיות שנייה, והתפוקה מגיעה עד 10,000 הודעות/שנייה. עיבדנו מעל 500 מיליון אירועים עבור לקוחות עם עקביות נתונים של 99.99%.

הגדרת Kafka Connect turnkey: 4 שלבים

בצע את השלבים הבאים כדי להגדיר את האינטגרציה שלך לסטרימינג. שלב 1: הכנת PostgreSQL ו-Kafka

הפעל שכפול לוגי: wal_level = logical, צור פרסום עבור הטבלאות הנדרשות ומשתמש debezium עם הרשאות SELECT. בצד Kafka, בדוק bootstrap.servers, הגדרות retention, ונושאים compact עבור Debezium.

שלב 2: פריסת Kafka Connect במצב מבוזר

קלאסטר של 2–3 צמתים עם נושאים פנימיים עבור תצורה. התצורה מוקלדת במלואה (ראה להלן). אנו משתמשים ב-Avro עם Schema Registry — זה מבטיח תאימות סכמה כאשר מסד הנתונים משתנה.

שלב 3: הגדרת Debezium Source Connector

Debezium קורא את PostgreSQL WAL ושולח כל שינוי ל-Kafka. הגדר snapshot.mode=initial, transforms לחילוץ השורה החדשה, ו-tombstone עבור DELETE. עבור טבלאות גדולות (מיליארדי שורות), השתמש ב-incremental snapshot כדי למנוע חסימת מסד הנתונים.

שלב 4: Sink connectors ל-Elasticsearch ו-PostgreSQL

עבור אינדקס החיפוש — Elasticsearch Sink עם אצווה של 500 רשומות ו-retry backoff. עבור מסד הנתונים האנליטי — JDBC Sink עם upsert ו-pk.mode=record_key. בכל שלב, בדוק INSERT/UPDATE/DELETE וחביון.

איך לפרוס Kafka Connect ו-connectors?

להלן התצורות המרכזיות להפעלת קלאסטר מבוזר ו-connectors טיפוסיים.

# distributed-свойства (connect-distributed.properties)
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
group.id=kafka-connect-cluster
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-statuses
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3
offset.flush.interval.ms=10000
rest.host.name=0.0.0.0
rest.port=8083
rest.advertised.host.name=connect-1.internal
rest.advertised.port=8083
plugin.path=/opt/kafka/plugins
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081

לפני הרצת Debezium, הגדר את PostgreSQL:

ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;
CREATE USER debezium WITH REPLICATION LOGIN PASSWORD 'secure_password';
GRANT CONNECT ON DATABASE myapp TO debezium;
GRANT USAGE ON SCHEMA public TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium;
CREATE PUBLICATION debezium_pub FOR TABLE products, orders, users, categories;

כעת רשום את ה-connectors דרך REST API. הנה דוגמה ל-Debezium Source ו-JDBC Sink בבלוק אחד:

# Debezium Source
curl -X POST http://connect-1:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "postgres-source-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres.internal",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "secure_password",
      "database.dbname": "myapp",
      "database.server.name": "myapp-pg",
      "topic.prefix": "myapp",
      "table.include.list": "public.products,public.orders,public.users",
      "plugin.name": "pgoutput",
      "publication.name": "debezium_pub",
      "slot.name": "debezium_slot",
      "snapshot.mode": "initial",
      "snapshot.isolation.mode": "read_committed",
      "decimal.handling.mode": "double",
      "time.precision.mode": "connect",
      "tombstones.on.delete": "true",
      "heartbeat.interval.ms": "10000",
      "transforms": "unwrap",
      "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
      "transforms.unwrap.delete.handling.mode": "rewrite",
      "transforms.unwrap.add.fields": "op,ts_ms,source.ts_ms"
    }
  }'

# JDBC Sink
curl -X POST http://connect-1:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "postgres-sink-connector",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
      "tasks.max": "4",
      "topics": "myapp.analytics.events",
      "connection.url": "jdbc:postgresql://analytics-pg:5432/analytics",
      "connection.user": "kafka_writer",
      "connection.password": "secure_password",
      "auto.create": "false",
      "auto.evolve": "false",
      "insert.mode": "upsert",
      "pk.mode": "record_key",
      "pk.fields": "id",
      "table.name.format": "analytics.${topic}",
      "batch.size": "1000",
      "db.timezone": "UTC",
      "transforms": "dropPrefix",
      "transforms.dropPrefix.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
      "transforms.dropPrefix.exclude": "__deleted,__op,__ts_ms"
    }
  }'

ניהול ה-connectors נעשה דרך REST API: שליפת סטטוס, השהיה, הפעלה מחדש של משימות שנכשלו. אנו משלבים את Kafka Connect עם ניטור Prometheus למדדים בזמן אמת. מדדי Prometheus JMX מוגדרים באמצעות JMX Exporter.

בעיות טיפוסיות ופתרונותיהן

מה כלול בעבודה

• ניתוח סכמת מסד הנתונים הנוכחית והעומסים, בחירת connectors וטרנספורמציות
• הגדרת PostgreSQL לשכפול לוגי (WAL, פרסום, משתמשים)
• התקנת Kafka Connect במצב מבוזר על 2–3 צמתים עם Schema Registry
• פריסת Debezium Source Connector עם snapshot ראשוני
• הגדרת אחד או יותר Sink connectors (Elasticsearch, JDBC, S3)
• כתיבת Single Message Transforms (SMT) ליישור סכמות
• שילוב ניטור: Prometheus JMX Exporter, לוח מחוונים Grafana, התראות Slack
• תיעוד סכמת הנושאים והתצורות
• הדרכת צוות (שעתיים: פעולות בסיסיות, הפעלה מחדש, אבחון)

לוח זמנים

במשך למעלה מ-10 שנים סיפקנו 50+ צינורות אינטגרציה על PostgreSQL, MySQL ו-MongoDB. קבל ייעוץ לפרויקט שלך – המהנדסים שלנו ינתחו את הסכמה והעומס שלך ויציעו את הארכיטקטורה האופטימלית. צור קשר — נעריך את הפרויקט שלך תוך שעתיים. הזמן הגדרת Kafka Connect — ואנו מבטיחים אספקת שינויים תוך שניות.