הכירו את Debezium - CDC עבור אפאצ'י קפקא

בעבודתי, אני נתקל לעתים קרובות בפתרונות טכניים / מוצרי תוכנה חדשים, שמידע עליהם די דל באינטרנט דובר הרוסית. במאמר זה, אנסה למלא פער אחד כזה עם דוגמה מהתרגול האחרון שלי, כאשר הייתי צריך להגדיר שליחת אירועי CDC משני DBMSs פופולריים (PostgreSQL ו-MongoDB) לאשכול קפקא באמצעות Debezium. אני מקווה שמאמר סקירה זה, שהופיע כתוצאה מהעבודה שנעשתה, יהיה שימושי לאחרים.

מה זה Debezium ו-CDC באופן כללי?

דביציום - נציג של קטגוריית התוכנה CDC (שינוי לכידת נתונים), או ליתר דיוק, זוהי קבוצה של מחברים עבור DBMSs שונים התואמים למסגרת Apache Kafka Connect.

זה פרויקט קוד פתוח, מורשה תחת רישיון Apache v2.0 ובחסות Red Hat. הפיתוח מתבצע מאז 2016 וכרגע הוא מספק תמיכה רשמית ל-DBMS הבאים: MySQL, PostgreSQL, MongoDB, SQL Server. ישנם גם מחברים לקסנדרה ואורקל, אך הם נמצאים כעת בסטטוס "גישה מוקדמת", ומהדורות חדשות אינן מבטיחות תאימות לאחור.

אם נשווה את CDC עם הגישה המסורתית (כשהאפליקציה קוראת נתונים מה-DBMS ישירות), אז היתרונות העיקריים שלה כוללים יישום הזרמת שינויי נתונים ברמת השורה עם חביון נמוך, אמינות וזמינות גבוהות. שתי הנקודות האחרונות מושגות על ידי שימוש באשכול קפקא כמאגר לאירועי CDC.

כמו כן, היתרונות כוללים את העובדה שמודל יחיד משמש לאחסון אירועים, כך שהאפליקציה הסופית לא צריכה לדאוג לניואנסים של הפעלת DBMS שונים.

לבסוף, שימוש במתווך הודעות פותח טווח לשינוי קנה מידה אופקי של יישומים העוקבים אחר שינויים בנתונים. יחד עם זאת, ההשפעה על מקור הנתונים מצטמצמת, מאחר שהנתונים מתקבלים לא ישירות מה-DBMS, אלא מאשכול קפקא.

על ארכיטקטורת דבזיום

השימוש ב-Debezium מסתכם בתכנית הפשוטה הזו:

DBMS (כמקור נתונים) → מחבר ב-Kafka Connect → Apache Kafka → צרכן

כהמחשה, אתן תרשים מאתר הפרויקט:

עם זאת, אני לא ממש אוהב את התוכנית הזו, כי נראה שרק מחבר כיור אפשרי.

במציאות, המצב שונה: מילוי Data Lake שלך (קישור אחרון בתרשים למעלה) זו לא הדרך היחידה להשתמש ב-Debezium. אירועים שנשלחים ל- Apache Kafka יכולים לשמש את האפליקציות שלך כדי להתמודד עם מצבים שונים. לדוגמה:

• הסרה של נתונים לא רלוונטיים מהמטמון;
• שליחת הודעות;
• עדכוני אינדקס חיפוש;
• איזשהו יומני ביקורת;
• ...

במידה ויש לכם אפליקציית ג'אווה ואין צורך/אפשרות להשתמש באשכול קפקא, ישנה גם אפשרות לעבוד דרך מחבר מוטבע. היתרון הברור הוא שבאמצעותו ניתן לסרב לתשתית נוספת (בצורת מחבר וקפקא). עם זאת, פתרון זה הוצא מגרסה 1.1 ואינו מומלץ עוד לשימוש (ייתכן שהוא יוסר במהדורות עתידיות).

מאמר זה ידון בארכיטקטורה המומלצת על ידי מפתחים, המספקת סובלנות תקלות ומדרגיות.

תצורת מחבר

על מנת להתחיל לעקוב אחר שינויים בערך החשוב ביותר - נתונים - אנו צריכים:

1. מקור נתונים, שיכול להיות MySQL החל מגרסה 5.7, PostgreSQL 9.6+, MongoDB 3.2+ (רשימה מלאה);
2. אשכול אפאצ'י קפקא
3. מופע Kafka Connect (גרסאות 1.x, 2.x);
4. מחבר Debezium מוגדר.

עבדו על שתי הנקודות הראשונות, כלומר. תהליך התקנת DBMS ו- Apache Kafka הם מעבר להיקף המאמר. עם זאת, למי שרוצה לפרוס הכל בארגז חול, יש אחד מוכן במאגר הרשמי עם דוגמאות docker-compose.yaml.

נתמקד בשתי הנקודות האחרונות ביתר פירוט.

0. קפקא קונקט

כאן ובהמשך המאמר, כל דוגמאות התצורה נלקחות בחשבון בהקשר של תמונת Docker המופצת על ידי מפתחי Debezium. הוא מכיל את כל קבצי הפלאגין הדרושים (מחברים) ומספק קפקא קונקט תצורה באמצעות משתני סביבה.

אם אתה מתכוון להשתמש ב-Kafka Connect מ-Confluent, תצטרך להוסיף את התוספים של המחברים הדרושים בעצמך לספרייה המצוינת ב- plugin.path או להגדיר באמצעות משתנה סביבה CLASSPATH. ההגדרות של העובד והמחברים של Kafka Connect מוגדרות באמצעות קבצי תצורה המועברים כארגומנטים לפקודת התחל של העובד. לפרטים ראה תיעוד.

כל תהליך הקמת Debeizum בגרסת המחבר מתבצע בשני שלבים. הבה נשקול כל אחד מהם:

1. הקמת המסגרת של Kafka Connect

כדי להזרים נתונים לאשכול Apache Kafka, פרמטרים ספציפיים מוגדרים במסגרת Kafka Connect, כגון:

• הגדרות חיבור אשכול,
• שמות של נושאים בהם תישמר תצורת המחבר עצמו,
• שם הקבוצה שבה פועל המחבר (במקרה של שימוש במצב מבוזר).

תמונת Docker הרשמית של הפרויקט תומכת בתצורה באמצעות משתני סביבה - זה מה שנשתמש. אז בוא נוריד את התמונה:

docker pull debezium/connect

קבוצת משתני הסביבה המינימלית הנדרשת להפעלת המחבר היא כדלקמן:

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 - רשימה ראשונית של שרתי אשכול קפקא כדי לקבל רשימה מלאה של חברי אשכול;
• OFFSET_STORAGE_TOPIC=connector-offsets - נושא לאחסון עמדות שבהן המחבר נמצא כעת;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status - נושא לאחסון מצב המחבר ומשימותיו;
• CONFIG_STORAGE_TOPIC=connector-config - נושא לאחסון נתוני תצורת מחברים ומשימותיו;
• GROUP_ID=1 - מזהה של קבוצת העובדים שעליה ניתן לבצע את משימת המחבר; נדרש בעת שימוש מבוזר (מופץ) מִשׁטָר.

אנו מתחילים את המיכל עם המשתנים הבאים:

docker run 
  -e BOOTSTRAP_SERVERS='kafka-1:9092,kafka-2:9092,kafka-3:9092' 
  -e GROUP_ID=1 
  -e CONFIG_STORAGE_TOPIC=my_connect_configs 
  -e OFFSET_STORAGE_TOPIC=my_connect_offsets 
  -e STATUS_STORAGE_TOPIC=my_connect_statuses  debezium/connect:1.2

הערה לגבי אברו

כברירת מחדל, Debezium כותב נתונים בפורמט JSON, שמקובל עבור ארגזי חול וכמויות קטנות של נתונים, אך יכול להוות בעיה במסדי נתונים עמוסים בכבדות. אלטרנטיבה לממיר JSON היא לבצע סדרה של הודעות באמצעות אברו לפורמט בינארי, המפחית את העומס על תת-מערכת ה-I/O ב- Apache Kafka.

כדי להשתמש ב- Avro, עליך לפרוס מכשיר נפרד schema-registry (לאחסון סכמות). המשתנים עבור הממיר ייראו כך:

name: CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: VALUE_CONVERTER   
value: io.confluent.connect.avro.AvroConverter

פרטים על השימוש ב-Avro והגדרת רישום עבורו הם מעבר לתחום המאמר - בהמשך, לשם הבהירות, נשתמש ב-JSON.

2. הגדרת המחבר עצמו

עכשיו אתה יכול ללכת ישירות לתצורת המחבר עצמו, שיקרא נתונים מהמקור.

בואו נסתכל על הדוגמה של מחברים לשני DBMS: PostgreSQL ו-MongoDB, שעבורם יש לי ניסיון ושיש ביניהם הבדלים (אמנם קטנים, אבל בחלק מהמקרים משמעותיים!).

התצורה מתוארת בסימון JSON ומועלת ל-Kafka Connect באמצעות בקשת POST.

2.1. PostgreSQL

תצורת מחבר לדוגמה עבור PostgreSQL:

{
  "name": "pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "127.0.0.1",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "definitelynotpassword",
    "database.dbname" : "dbname",
    "database.server.name": "pg-dev",
    "table.include.list": "public.(.*)",
    "heartbeat.interval.ms": "5000",
    "slot.name": "dbname_debezium",
    "publication.name": "dbname_publication",
    "transforms": "AddPrefix",
    "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.AddPrefix.regex": "pg-dev.public.(.*)",
    "transforms.AddPrefix.replacement": "data.cdc.dbname"
  }
}

עקרון הפעולה של המחבר לאחר תצורה זו הוא די פשוט:

• בהתחלה הראשונה הוא מתחבר למסד הנתונים שצוין בתצורה ומתחיל במצב תמונת מצב ראשונית, שליחת לקפקא את סט הנתונים הראשוני שהתקבל עם התנאי SELECT * FROM table_name.
• לאחר השלמת האתחול, המחבר נכנס למצב של קריאת שינויים מקבצי PostgreSQL WAL.

לגבי האפשרויות בהן נעשה שימוש:

• name - שם המחבר שעבורו משמשת התצורה המתוארת להלן; בעתיד, השם הזה ישמש לעבודה עם המחבר (כלומר להציג את הסטטוס / הפעל מחדש / עדכן את התצורה) דרך Kafka Connect REST API;
• connector.class - מחלקת מחברי DBMS שתשמש את המחבר המוגדר;
• plugin.name הוא שם התוסף לפענוח לוגי של נתונים מקבצי WAL. זמין לבחירה wal2json, decoderbuffs и pgoutput. שני הראשונים דורשים התקנה של ההרחבות המתאימות ב-DBMS, ו pgoutput עבור PostgreSQL גרסה 10 ומעלה אינו דורש מניפולציות נוספות;
• database.* — אפשרויות לחיבור למסד הנתונים, היכן database.server.name - השם של מופע PostgreSQL המשמש ליצירת שם הנושא באשכול קפקא;
• table.include.list - רשימה של טבלאות שבהן אנו רוצים לעקוב אחר שינויים; ניתן בפורמט schema.table_name; לא ניתן להשתמש יחד עם table.exclude.list;
• heartbeat.interval.ms - מרווח (במילישניות) שבו המחבר שולח הודעות פעימות לב לנושא מיוחד;
• heartbeat.action.query - בקשה שתבוצע בעת שליחת כל הודעת דופק (האפשרות הופיעה מאז גרסה 1.1);
• slot.name - שם חריץ השכפול שישמש את המחבר;
• publication.name - שם פרסום ב-PostgreSQL שהמחבר משתמש בו. במקרה שהוא לא קיים, Debezium ינסה ליצור אותו. אם למשתמש שתחתיו נעשה החיבור אין מספיק זכויות לפעולה זו, המחבר ייצא עם שגיאה;
• transforms קובע כיצד בדיוק לשנות את השם של נושא היעד:
  • transforms.AddPrefix.type מציין שנשתמש בביטויים רגולריים;
  • transforms.AddPrefix.regex - מסכה שלפיה שמו של נושא היעד מוגדר מחדש;
  • transforms.AddPrefix.replacement - ישירות מה אנחנו מגדירים מחדש.

עוד על פעימות לב ושינויים

כברירת מחדל, המחבר שולח נתונים לקפקא עבור כל עסקה שהתחייבה, וכותב את ה-LSN שלו (מספר רצף יומן) לנושא השירות offset. אבל מה קורה אם המחבר מוגדר לקרוא לא את כל מסד הנתונים, אלא רק חלק מהטבלאות שלו (שבהן הנתונים מתעדכנים לעתים רחוקות)?

• המחבר יקרא קבצי WAL ולא יזהה בהם התחייבויות עסקאות לטבלאות שהוא מנטר.
• לכן, הוא לא יעדכן את מיקומו הנוכחי לא בנושא ולא במשבצת השכפול.
• זה, בתורו, יגרום לקבצי WAL להיות "תקועים" בדיסק וככל הנראה יגמר שטח הדיסק.

וכאן האפשרויות באות לעזרה. heartbeat.interval.ms и heartbeat.action.query. שימוש באפשרויות אלו בזוגות מאפשר לבצע בקשה לשינוי נתונים בטבלה נפרדת בכל פעם שנשלחת הודעת דופק. לפיכך, ה-LSN עליו נמצא המחבר כרגע (בחריץ השכפול) מתעדכן כל הזמן. זה מאפשר ל-DBMS להסיר קבצי WAL שאינם נחוצים עוד. למידע נוסף על אופן פעולת האפשרויות, ראה תיעוד.

אפשרות נוספת שראויה לתשומת לב רבה יותר היא transforms. למרות שזה יותר על נוחות ויופי...

כברירת מחדל, Debezium יוצר נושאים באמצעות מדיניות השמות הבאה: serverName.schemaName.tableName. זה אולי לא תמיד נוח. אפשרויות transforms באמצעות ביטויים רגולריים, ניתן להגדיר רשימה של טבלאות שיש לנתב את האירועים שלהן לנושא בעל שם ספציפי.

בתצורה שלנו הודות ל transforms קורה הבא: כל אירועי ה-CDC ממסד הנתונים במעקב יעברו לנושא עם השם data.cdc.dbname. אחרת (ללא הגדרות אלה), Debezium כברירת מחדל תיצור נושא עבור כל טבלה בטופס: pg-dev.public.<table_name>.

מגבלות מחברים

בסוף התיאור של תצורת המחבר עבור PostgreSQL, כדאי לדבר על התכונות/מגבלות הבאות של עבודתו:

1. פונקציונליות המחבר עבור PostgreSQL מסתמכת על הרעיון של פענוח לוגי. לכן הוא אינו עוקב אחר בקשות לשינוי מבנה מסד הנתונים (DDL) - בהתאם, נתונים אלו לא יהיו בנושאים.
2. מכיוון שמשתמשים בחריצי שכפול, החיבור של המחבר אפשרי רק למופע הראשי של ה-DBMS.
3. אם למשתמש שתחתיו המחבר מתחבר למסד הנתונים יש זכויות קריאה בלבד, אז לפני ההשקה הראשונה, יהיה עליך ליצור ידנית משבצת שכפול ולפרסם במסד הנתונים.

החלת תצורה

אז בואו נטען את התצורה שלנו למחבר:

curl -i -X POST -H "Accept:application/json" 
  -H  "Content-Type:application/json"  http://localhost:8083/connectors/ 
  -d @pg-con.json

אנו בודקים שההורדה הצליחה והמחבר התחיל:

$ curl -i http://localhost:8083/connectors/pg-connector/status 
HTTP/1.1 200 OK
Date: Thu, 17 Sep 2020 20:19:40 GMT
Content-Type: application/json
Content-Length: 175
Server: Jetty(9.4.20.v20190813)

{"name":"pg-connector","connector":{"state":"RUNNING","worker_id":"172.24.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.24.0.5:8083"}],"type":"source"}

נהדר: זה מוגדר ומוכן לשימוש. עכשיו בואו נתחזה לצרכן ונתחבר לקפקא, ולאחר מכן נוסיף ונשנה ערך בטבלה:

$ kafka/bin/kafka-console-consumer.sh 
  --bootstrap-server kafka:9092 
  --from-beginning 
  --property print.key=true 
  --topic data.cdc.dbname

postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', '[email protected]');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1

בנושא שלנו, זה יוצג באופן הבא:

JSON ארוך מאוד עם השינויים שלנו

{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":null,
"after":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600374991648,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":602,
"lsn":34088472,
"xmin":null
},
"op":"c",
"ts_ms":1600374991762,
"transaction":null
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"after":{
"id":1005,
"first_name":"egg",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600375609365,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":603,
"lsn":34089688,
"xmin":null
},
"op":"u",
"ts_ms":1600375609778,
"transaction":null
}
}

בשני המקרים, הרשומות מורכבות מהמפתח (PK) של הרשומה ששונתה, ומעצם השינויים: מה הייתה הרשומה לפני ומה היא הפכה לאחריה.

• במקרה של INSERT: ערך לפני (before) שווים nullואחריו המחרוזת שהוכנסה.
• במקרה של UPDATE: ב payload.before המצב הקודם של השורה מוצג, ונכנס payload.after - חדש עם מהות השינוי.

2.2 MongoDB

מחבר זה משתמש במנגנון השכפול הסטנדרטי של MongoDB, קורא מידע מה-oplog של הצומת הראשי של DBMS.

בדומה למחבר שתואר כבר עבור PgSQL, גם כאן, בהתחלה הראשונה, נלקחת תמונת המצב הראשית של הנתונים, ולאחר מכן המחבר עובר למצב קריאת oplog.

דוגמה לתצורה:

{
"name": "mp-k8s-mongo-connector",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"tasks.max": "1",
"mongodb.hosts": "MainRepSet/mongo:27017",
"mongodb.name": "mongo",
"mongodb.user": "debezium",
"mongodb.password": "dbname",
"database.whitelist": "db_1,db_2",
"transforms": "AddPrefix",
"transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.AddPrefix.regex": "mongo.([a-zA-Z_0-9]*).([a-zA-Z_0-9]*)",
"transforms.AddPrefix.replacement": "data.cdc.mongo_$1"
}
}

כפי שניתן לראות, אין אפשרויות חדשות לעומת הדוגמה הקודמת, אלא רק מספר האפשרויות האחראיות לחיבור למסד הנתונים והקידומות שלהן צומצם.

הגדרות transforms הפעם הם עושים את הפעולות הבאות: מפנים את השם של נושא היעד מהתכנית <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

סובלנות לתקלות

סוגיית סובלנות התקלות והזמינות הגבוהה בזמננו היא חריפה מתמיד – במיוחד כשמדברים על נתונים ועסקאות, ומעקב אחר שינויי נתונים לא נמצא בצד בעניין הזה. בואו נסתכל מה יכול להשתבש עקרונית ומה יקרה לדבזיום בכל מקרה.

ישנן שלוש אפשרויות לביטול הסכמה:

1. כשל של קפקא קונקט. אם Connect מוגדר לעבוד במצב מבוזר, זה דורש ממספר עובדים להגדיר את אותו group.id. לאחר מכן, אם אחד מהם נכשל, המחבר יופעל מחדש על העובד השני וימשיך לקרוא מהעמדה המחוייבת האחרונה בנושא בקפקא.
2. אובדן קישוריות עם אשכול קפקא. המחבר פשוט יפסיק לקרוא במיקום שלא הצליח לשלוח לקפקא ומדי פעם ינסה לשלוח אותו מחדש עד שהניסיון יצליח.
3. מקור הנתונים אינו זמין. המחבר ינסה להתחבר מחדש למקור בהתאם לתצורה. ברירת המחדל היא 16 ניסיונות שימוש גיבוי אקספוננציאלי. לאחר הניסיון ה-16 הכושל, המשימה תסומן כ נכשל ויהיה צורך להפעיל אותו מחדש באופן ידני באמצעות ממשק Kafka Connect REST.
   • במקרה של PostgreSQL נתונים לא יאבדו, כי שימוש בחריצי שכפול ימנע מחיקה של קבצי WAL שלא נקראו על ידי המחבר. במקרה זה, יש חיסרון: אם קישוריות הרשת בין המחבר ל-DBMS מופרעת במשך זמן רב, יש סיכוי ששטח הדיסק יגמר, וזה עלול להוביל לכשל של ה-DBMS כולו.
   • במקרה של MySQL ניתן לסובב קבצי binlog על ידי ה-DBMS עצמו לפני שחזור הקישוריות. זה יגרום למחבר לעבור למצב כשל, והוא יצטרך להפעיל מחדש במצב תמונת מצב ראשוני כדי להמשיך לקרוא מ-binlogs כדי לשחזר את הפעולה הרגילה.
   • על MongoDB. התיעוד אומר: ההתנהגות של המחבר במקרה שקובצי היומן/אופלוגים נמחקו והמחבר לא יכול להמשיך לקרוא מהמיקום שבו הופסק זהה עבור כל ה-DBMS. זה טמון בעובדה שהמחבר ייכנס למדינה נכשל וידרוש הפעלה מחדש במצב תמונת מצב ראשונית.

     עם זאת, ישנם יוצאי דופן. אם המחבר היה במצב מנותק במשך זמן רב (או לא הצליח להגיע למופע MongoDB), וה-oplog הסתובב במהלך זמן זה, אז כאשר החיבור ישוחזר, המחבר ימשיך ברוגע לקרוא נתונים מהמיקום הזמין הראשון , ולכן חלק מהנתונים בקפקא לא יפגע.

מסקנה

Debezium הוא הניסיון הראשון שלי עם מערכות CDC והיה מאוד חיובי בסך הכל. הפרויקט שיחד את תמיכת ה-DBMS הראשי, קלות תצורה, תמיכה באשכולות וקהילה פעילה. למעוניינים בתרגול, אני ממליץ לקרוא את המדריכים עבור קפקא מתחבר и דביציום.

בהשוואה למחבר JDBC עבור Kafka Connect, היתרון העיקרי של Debezium הוא ששינויים נקראים מיומני ה-DBMS, מה שמאפשר קבלת נתונים בהשהייה מינימלית. ה-JDBC Connector (שמסופק על ידי Kafka Connect) מבצע שאילתות על הטבלה המנוקבת במרווח קבוע ו(מאותה סיבה) אינו מייצר הודעות כאשר הנתונים נמחקים (איך ניתן לבצע שאילתה לנתונים שאינם קיימים?).

כדי לפתור בעיות דומות, אתה יכול לשים לב לפתרונות הבאים (בנוסף ל-Debezium):

• מחבר JDBC Kafka Connect
• כמה פתרונות MySQL בלבד:
  • MySQL Streamer;
  • מקסוול;
  • Spinal Tap;
• אורקל גולדן גייט, אבל זו "קטגוריית משקל" שונה לחלוטין.

נ.ב.

קרא גם בבלוג שלנו:

• «קבע את הגודל המתאים לאשכול קפקא ב-Kubernetes";
• «סיפורים מעשיים מחיי היומיום שלנו ב-SRE. חלק 2";
• «סקירה קצרה של מפעילי PostgreSQL עבור Kubernetes, הבחירה והניסיון שלנו".

מקור: www.habr.com