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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

חשבון השירות של Managed Kafka הוא בפורמט הבא: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, כאשר PROJECT_NUMBER הוא מספר הפרויקט של אשכול Connect.

אם אשכול 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. בשדה Value, בוחרים באפשרות 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. בחלונית הניווט, לוחצים על משתמשים.

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

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

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

    service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com

    מחליפים את 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.

בגלל מגבלת האורך של שם משתמש במסד נתונים, הסיומת .gserviceaccount.com מושמטת משם המשתמש, כך ששם המשתמש הוא 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 של נקודת הקצה ל- דומיינים של DNS שניתנים לפתרון של אשכול 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

  1. מפעילים את Cloud Shell במסוף Google Cloud .

    הפעלת Cloud Shell

    בחלק התחתון של מסוף Google Cloud יתחיל סשן של Cloud Shell ותופיע הודעה של שורת הפקודה. Cloud Shell היא סביבת מעטפת שבה ה-CLI של Google Cloud מותקן ומוגדרים ערכים לפרויקט הקיים. הסשן יופעל תוך כמה שניות.

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

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

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

    דוגמה לקובץ תצורה של המחבר למקור 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 או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.