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

במאמר הזה מוסבר איך ליצור מחבר מקור של Cloud SQL ל-PostgreSQL ל-Kafka Connect.

מחבר של מקור Cloud SQL ל-PostgreSQL הוא מופע של מחבר Debezium 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 Client‏ (roles/cloudsql.client)
  • משתמש במופע Cloud SQL (roles/cloudsql.instanceUser)

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

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

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

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

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

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

המסוף

  1. עוברים אל Cloud SQL > Instances (מכונות).

    כניסה לדף Instances

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

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

  4. מרחיבים את הקטע Flags and parameters (דגלים ופרמטרים).

  5. לוחצים על הוספת דגל מסד נתונים.

  6. ברשימה Choose a flag, בוחרים באפשרות 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, מוסיפים את חשבון השירות של 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 של נקודת הקצה (endpoint) של Private Service Connect. מידע נוסף זמין במאמר התחברות למופע באמצעות Private Service Connect.
  2. מוסיפים את שם ה-DNS של נקודת הקצה אל Resolvable DNS domains (דומייני DNS שאפשר לפתור) של אשכול Connect. מידע נוסף זמין במאמר עדכון של אשכול Connect.
כתובת IP ציבורית
  1. מגדירים כתובת IP ציבורית למופע. מידע נוסף מופיע במאמר בנושא הגדרת כתובת IP ציבורית.
  2. מגדירים NAT ציבורי כדי לאפשר לעובדי אשכול Connect לתקשר עם האינטרנט. מידע נוסף זמין במאמר בנושא הגדרת NAT ציבורי. כשיוצרים את שער Cloud NAT, צריך לציין את רשת ה-VPC שמכילה את תת-הרשת הראשית של אשכול Connect.

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

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

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

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

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

המסוף

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

    מעבר אל Connect Clusters

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

  3. לוחצים על יצירת מחבר.

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

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

  5. בקטע Connector plugin (תוסף מחבר), בוחרים באפשרות Cloud SQL for PostgreSQL Source (מקור Cloud SQL ל-PostgreSQL).

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

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

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

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

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

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

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

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

gcloud

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

    הפעלת 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

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

    • ‫CONNECTOR_ID: המזהה או השם של המחבר. הנחיות לשמות של מחברים זמינות במאמר הנחיות לשמות של משאבים בשירות מנוהל ל-Apache Kafka. אי אפשר לשנות את השם של מחבר.

    • ‫LOCATION: המיקום שבו יוצרים את מחבר. המיקום הזה צריך להיות זהה למיקום שבו יצרתם את אשכול Connect.

    • ‫CONNECT_CLUSTER_ID: המזהה של אשכול Connect שבו נוצר המחבר.

    • ‫CONFIG_FILE: הנתיב לקובץ ההגדרות של המחבר בפורמט YAML.

    דוגמה לקובץ הגדרה של מחבר המקור 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.

מידע נוסף זמין במאמר שמות של נושאים במסמכי התיעוד של Debezium.

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