יצירת מחבר של מקור Cloud SQL ל-PostgreSQL

מחבר מקור של Cloud SQL ל-PostgreSQL מאפשר לתעד שינויים ברמת השורה ממסד נתונים של Cloud SQL ל-PostgreSQL ולכתוב אותם לנושאים באשכול של שירות מנוהל ל-Apache Kafka.

תרחישים נפוצים לדוגמה לשימוש במחבר הזה:

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

לפני שמתחילים

לפני שיוצרים מחבר מקור של Cloud SQL ל-PostgreSQL, צריך לוודא שיש לכם:

תפקידים והרשאות נדרשים

כדי לקבל את ההרשאות שנדרשות ליצירת מחבר, צריך לבקש מהאדמין להקצות לכם ב-IAM את התפקיד עריכת מחבר מנוהל של Kafka (roles/managedkafka.connectorEditor) בפרויקט. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

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

ההרשאות הנדרשות

כדי ליצור מחבר, צריך את ההרשאות הבאות:

  • יוצרים מחבר: managedkafka.connectors.create

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

מתן הרשאות לקריאה מ-Cloud SQL

לסוכן השירות המנוהל של Kafka צריכה להיות הרשאת גישה ל-Cloud SQL ל-PostgreSQL. מקצים לסוכן השירות את תפקידי ה-IAM הבאים:

  • לקוח Cloud SQL‏ (roles/cloudsql.client)
  • משתמש במכונה של Cloud SQL‏ (roles/cloudsql.instanceUser)

אם אשכול Connect נמצא בפרויקט אחר מאשכול השירות המנוהל ל-Apache Kafka, קראו את המאמר יצירת אשכול Connect בפרויקט אחר.

הגדרת מסד הנתונים

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

הפעלת פענוח לוגי

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

המסוף

  1. עוברים אל Cloud SQL > Instances.

    כניסה לדף Instances

  2. לוחצים על השם של המכונה.

  3. לוחצים על עריכה.

  4. מרחיבים את הקטע דגלים ופרמטרים.

  5. לוחצים על Add a database flag.

  6. ברשימה בחירת תווית, בוחרים באפשרות cloudsql.logical_decoding.

  7. בשדה ערך, בוחרים באפשרות On.

  8. לוחצים על סיום.

  9. לוחצים על Save.

מידע נוסף זמין במאמר הגדרה של שכפול לוגי ופענוח.

הגדרת סימון נתונים שהשתנו (CDC)

אחרי שמפעילים פענוח לוגי במכונה, מפעילים את התכונה "סימון נתונים שהשתנו" (CDC) בטבלאות שרוצים לשכפל.

כדי להפעיל CDC בטבלה, מריצים את הצהרת ה-SQL‏ CREATE PUBLICATION. ההצהרה הזו יוצרת פרסום, שמגדיר קבוצת טבלאות לשכפול.

  • אפשרות 1. יוצרים פרסום שמשכפל את השינויים בכל הטבלאות במסד הנתונים.

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

  • אפשרות 2. יוצרים פרסום לקבוצה ספציפית של טבלאות.

    CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;

    מחליפים את TABLE_LIST ברשימה מופרדת בפסיקים של טבלאות, בפורמט "schema_name"."table_name". כדי למנוע שגיאות תחביר אם השמות מכילים תווים מיוחדים או אותיות רישיות, צריך להקיף את שמות הסכימה והטבלה במירכאות כפולות, כמו בדוגמה.

כברירת מחדל, המחבר משתמש ב-dbz_publication בתור שם אתר החדשות. כדי להשתמש באתר חדשות עם שם אחר, קראו את המאמר שם אתר החדשות.

יצירת חשבון משתמש לחשבון השירות של Managed Kafka

המחבר של מקור Cloud SQL ל-PostgreSQL משתמש באימות של מסד נתונים ב-IAM כדי להתחבר למסד הנתונים. כדי להפעיל אימות של מסד נתונים ב-IAM, מוסיפים את חשבון השירות של Managed Kafka למכונה של Cloud SQL, באופן הבא:

המסוף

  1. עוברים אל Cloud SQL > Instances.

    כניסה לדף Instances

  2. לוחצים על השם של המכונה.

  3. בחלונית הניווט, לוחצים על Users.

  4. לוחצים על הוספת חשבון משתמש.

  5. בחלונית Add a user account, בוחרים באפשרות Cloud IAM.

  6. בשדה IAM principal, מזינים את הערך הבא:

    service-PROJECT_NUMBER@gcp-sa-managedkafka.

    מחליפים את PROJECT_NUMBER במספר הפרויקט של אשכול Connect.

  7. לוחצים על הוספה.

gcloud

מריצים את הפקודה gcloud sql users create:

gcloud sql users create service-PROJECT_NUMBER@gcp-sa-managedkafka.iam \
  --instance=INSTANCE_NAME \
  --type=cloud_iam_service_account

מחליפים את מה שכתוב בשדות הבאים:

  • ‫PROJECT_NUMBER: מספר הפרויקט של האשכול ב-Connect.

  • ‫INSTANCE_NAME: השם של המכונה של Cloud SQL ל-PostgreSQL.

בגלל מגבלת האורך של שם משתמש במסד נתונים, הסיומת . מושמטת משם המשתמש, כך ששם המשתמש הוא service-PROJECT_NUMBER@gcp-sa-managedkafka.iam. כשמריצים שאילתות SQL שמפנות לחשבון המשתמש ב-IAM, צריך לציין את השם המקוצר.

הגדרת חשבון המשתמש

אחרי שיוצרים את חשבון המשתמש ב-IAM, מתחברים למסד הנתונים כמשתמש עם התפקיד cloudsqlsuperuser (למשל המשתמש postgres שמוגדר כברירת מחדל), ומריצים את שאילתות ה-SQL הבאות.

המסוף

  1. מאפשרים למשתמש לקרוא את יומן הכתיבה מראש.

    ALTER USER "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam" WITH REPLICATION;

  2. נותנים למשתמש את ההרשאה SELECT בטבלאות.

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME"
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

    לחלופין, אפשר לתת את ההרשאה SELECT בטבלאות ספציפיות. אם בוחרים באפשרות הזו, צריך גם להגדיר את מאפיין ההגדרה table.include.list של המחבר לרשימת הטבלאות המותרות. שאילתת ה-SQL הבאה מעניקה הרשאת SELECT בטבלה אחת:

    GRANT SELECT ON TABLE SCHEMA_NAME.TABLE_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

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

    GRANT USAGE ON SCHEMA SCHEMA_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

הגדרה של רשתות

אפשר לחבר מחבר מקור של Cloud SQL ל-PostgreSQL למכונה של Cloud SQL בדרכים הבאות:

  • כתובת IP פרטית
  • התחברות לשירות פרטי
  • כתובת IP ציבורית

למידע נוסף על האפשרויות האלה, ראו בחירת אופן ההתחברות ל-Cloud SQL. כדי לשמור על אבטחה, מומלץ להשתמש בכתובת IP פרטית או ב-Private Service Connect, כי האפשרויות האלה לא דורשות חיבור לכתובת IP חיצונית.

בטבלה הבאה מפורטות דרישות הרשת לכל אפשרות:

סוג כתובת ה-IP דרישות
כתובת IP פרטית מגדירים כתובת IP פרטית למכונה. מידע נוסף זמין במאמר הגדרת כתובת IP פרטית.
התחברות לשירות פרטי
  1. מגדירים את Private Service Connect למכונה ומקבלים את שם ה-DNS של נקודת הקצה של Private Service Connect. מידע נוסף זמין במאמר התחברות למכונה באמצעות Private Service Connect.
  2. מוסיפים את שם ה-DNS של נקודת הקצה ל- Resolvable DNS domains של אשכול Connect. מידע נוסף זמין במאמר עדכון אשכול Connect.
כתובת IP ציבורית
  1. מגדירים כתובת IP ציבורית למכונה. מידע נוסף מופיע במאמר הגדרת כתובת IP ציבורית.
  2. מגדירים NAT ציבורי כדי לאפשר ל-workers של אשכול Connect לתקשר עם האינטרנט. מידע נוסף זמין במאמר הגדרת Public NAT. כשיוצרים את שער Cloud NAT, צריך לציין את רשת ה-VPC שמכילה את תת-הרשת הראשית של אשכול Connect.

יצירת מחבר מקור של Cloud SQL ל-PostgreSQL

כדי ליצור מחבר מקור של Cloud SQL ל-PostgreSQL, מבצעים את הפעולות הבאות:

כשהמחבר מופעל, הוא מבצע את הפעולות הבאות:

  1. יוצר תמונת מצב ראשונית של מסד הנתונים.
  2. יוצר נושא Kafka לכל טבלה שיש בה שורות.
  3. לכל שורה במסד הנתונים, נשלח אירוע שינוי לנושא המתאים.

בזמן שהמחבר פועל, הוא ממשיך לשלוח אירועי שינוי לנושאים. מידע נוסף על קובץ ה-snapshot הראשוני זמין בקטע Snapshots במסמכי התיעוד של Debezium.

המסוף

  1. נכנסים לדף Connect Clusters במסוף Google Cloud .

    מעבר אל Connect Clusters

  2. לוחצים על אשכול Connect שבו רוצים ליצור את המחבר.

  3. לוחצים על Create connector.

  4. מזינים מחרוזת בשם המחבר.

    הנחיות למתן שמות למשאבים של שירות מנוהל ל-Apache Kafka

  5. בשדה Connector plugin, בוחרים באפשרות Cloud SQL ל-PostgreSQL Source.

  6. ברשימה Instance, בוחרים את המכונה של Cloud SQL.

  7. ברשימה Database, בוחרים את מסד הנתונים של Cloud SQL.

  8. בשדה Topic prefix, מזינים קידומת לשמות הנושאים של Kafka. צריך לבחור קידומת ייחודית לכל מחבר מקור של Cloud SQL ל-PostgreSQL.

  9. אופציונלי: בשדה Table names, מזינים רשימה מופרדת בפסיקים של טבלאות, כדי לקרוא מהן את נתוני השינויים, בפורמט "schema_name"."table_name". אם משאירים את השדה הזה ריק, המחבר קורא את נתוני השינויים מכל הטבלאות שהן לא טבלאות מערכת במסד הנתונים.

  10. אופציונלי: בתיבה Configurations, מוסיפים מאפייני הגדרות או עורכים את מאפייני ברירת המחדל. מידע נוסף זמין במאמר הגדרת המחבר.

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

  11. אופציונלי: בוחרים את מדיניות ההפעלה מחדש של המשימה. למידע נוסף, קראו את המאמר מדיניות ההפעלה מחדש של משימות.

  12. לוחצים על יצירה.

gcloud

מריצים את הפקודה gcloud managed-kafka connectors create:

gcloud managed-kafka connectors create CONNECTOR_ID \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --config-file=CONFIG_FILE

מחליפים את מה שכתוב בשדות הבאים:

  • ‫CONNECTOR_ID: המזהה או השם של המחבר.
  • ‫CONNECT_CLUSTER_ID: המזהה של אשכול Connect שבו נוצר המחבר.
  • ‫LOCATION: המיקום של אשכול Connect.
  • ‫CONFIG_FILE: הנתיב לקובץ תצורה מקומי בפורמט YAML או JSON.

דוגמה לקובץ תצורה של מחבר המקור Cloud SQL ל-PostgreSQL:

connector.class: io.debezium.connector.postgresql.PostgresConnector
database.dbname: DATABASE_NAME
driver.cloudSqlInstance: INSTANCE_ID
driver.enableIamAuth: "true"
driver.ipTypes: IP_TYPES
driver.sslmode: disable
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: "false"
plugin.name: pgoutput
slot.name: SLOT_NAME
table.include.list: TABLE_LIST
topic.prefix: TOPIC_PREFIX
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: "true"

מחליפים את מה שכתוב בשדות הבאים:

  • ‫INSTANCE_ID: המזהה של המכונה ב-Cloud SQL שמכילה את מסד הנתונים, בפורמט הבא:

    PROJECT_ID:REGION:INSTANCE_NAME

  • ‫DATABASE_NAME: השם של מסד הנתונים ב-Cloud SQL שממנו רוצים לקרוא.

  • ‫IP_TYPES: רשימה מופרדת בפסיקים של סוגי כתובות IP

  • ‫SLOT_NAME: השם של משבצת הרפליקציה שרוצים ליצור.

  • ‫TABLE_LIST: רשימה מופרדת בפסיקים של טבלאות שמהן צריך לקרוא את נתוני השינויים, בפורמט "schema_name"."table_name".

  • ‫TOPIC_PREFIX: תחילית לשמות של נושאי Kafka.

הגדרת המחבר

בקטע הזה מתוארים כמה מאפייני הגדרה שאפשר להגדיר במחבר. הרשימה המלאה מופיעה במאמר Debezium connector for PostgreSQL במסמכי התיעוד של Debezium.

סוגים של כתובות IP

המאפיין driver.ipTypes מציין את סוג כתובת ה-IP שבה המחבר משתמש כדי להתחבר למסד הנתונים:

  • ‫PRIVATE: כתובת IP פרטית
  • PSC: Private Service Connect
  • ‫PUBLIC: כתובת IP ציבורית

המאפיין driver.ipTypes מכיל רשימה מופרדת בפסיקים של סוגי IP בסדר מועדף, לדוגמה, driver.ipTypes=PRIVATE,PUBLIC.

מידע נוסף זמין במאמר הגדרת רשת.

שם אתר החדשות

כברירת מחדל, המחבר מנסה להעביר בסטרימינג מאתר חדשות בשם dbz_publication. כדי לציין פרסום אחר, מוסיפים את publication.name=PUBLICATION_NAME להגדרה, כאשר PUBLICATION_NAME הוא שם הפרסום. לדוגמה: publication.name=my_publication.

משבצות שכפול

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

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

כברירת מחדל, המחבר מגדיר את המאפיין slot.drop.on.stop לערך false כדי למנוע אובדן נתונים. כשמוחקים מחבר באופן סופי, צריך למחוק ידנית את משבצת השכפול שבה המחבר השתמש. שם ברירת המחדל של משבצת הרפליקציה הוא debezium, אלא אם מגדירים אותו אחרת באמצעות המאפיין slot.name.

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

מסנן טבלה

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

  • ‫schema.include.list. רשימת הסכימות שרוצים לכלול.
  • ‫schema.exclude.list. רשימה של סכימות שצריך להחריג. אי אפשר להשתמש בו עם schema.include.list.
  • ‫table.include.list. רשימת הטבלאות שרוצים לכלול.
  • ‫table.exclude.list. רשימת הטבלאות שרוצים להחריג. אי אפשר להשתמש בו עם table.include.list.

שמות הנושאים

כברירת מחדל, המחבר יוצר נושאי Kafka לפי המוסכמה הבאה למתן שמות: topic_prefix.schema.table_name, כאשר topic.prefix הוא הערך של ההגדרה topic.prefix.

למידע נוסף, קראו את הקטע Topic names במסמכי התיעוד של Debezium.

המאמרים הבאים

‫Apache Kafka®‎ הוא סימן מסחרי רשום של The Apache Software Foundation או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.