יצירת אשכול Connect

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

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

מוודאים שכבר יצרתם אשכול של שירות מנוהל ל-Apache Kafka. צריך את השם של אשכול השירות המנוהל ל-Apache Kafka שאליו יצורף אשכול Connect.

כל אשכול Connect משויך לאשכול של שירות מנוהל ל-Apache Kafka. באשכול הזה מאוחסן המצב של המחברים שפועלים באשכול Connect.

תפקידים והרשאות שנדרשים כדי ליצור אשכול Connect

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

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

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

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

  • נותנים את ההרשאה ליצור אשכול Connect במיקום שצוין: managedkafka.connectClusters.create

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

גורמים ראשיים נדרשים ב-ACL

כברירת מחדל, אשכולות של שירות מנוהל ל-Apache Kafka מאפשרים גישה של Connect Cluster למשאבים אם לא מוגדרות רשימות ACL. כדי לעשות זאת, מגדירים את allow.everyone.if.no.acl.found ל-true, שהיא הגדרת ברירת המחדל.

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

חשבון השירות של אשכול Connect שמשמש כגורם המרכזי ברשימות בקרת גישה (ACL) הוא בפורמט הבא: User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com.

אם הגדרתם רשימות ACL באשכול Kafka, צריך להעניק לאשכול Connect הרשאות קריאה וכתיבה לנושאים והרשאות קריאה לקבוצות צרכנים באמצעות הפקודות הבאות:

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --operation WRITE --topic *

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --group *

מידע נוסף על הפקודות האלה זמין במאמר הגדרת רשימות ACL של Apache Kafka לבקרת גישה גרנולרית.

יצירת אשכול Connect בפרויקט אחר

בשירות המנוהל ל-Apache Kafka נעשה שימוש בסוכן שירות כדי לגשת למשאביGoogle Cloud . סוכן השירות משויך לפרויקט שבו יוצרים את האשכול.

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

כדי לתת את ההרשאות הנדרשות, צריך להקצות את התפקיד Managed Kafka Service Agent לסוכן השירות של אשכול Connect בפרויקט של אשכול Kafka. לדוגמה, אם יוצרים אשכול Kafka בפרויקט kafka-project ואשכול Connect בפרויקט connect-project, צריך להעניק את התפקיד 'סוכן שירות מנוהל של Kafka' בפרויקט kafka-project לסוכן השירות שמשויך לפרויקט connect-project.

כתובת האימייל של סוכן השירות היא בפורמט הבא: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, כאשר PROJECT_NUMBER הוא מספר הפרויקט. מידע נוסף על מתן התפקיד זמין במאמר איך יוצרים סוכני שירות ומעניקים להם תפקידים.

מאפיינים של אשכול Connect

בקטע הזה מוסבר על המאפיינים של אשכול Connect.

שם האשכול לחיבור

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

אשכול Kafka ראשי

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

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

אחרי שיוצרים את אשכול Connect, אי אפשר לעדכן אותו לאשכול Kafka אחר.

היתרונות של מיקום משותף באזור מבחינת זמן האחזור ועלויות הרשת

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

שיקולי עלות וזמן אחזור במערכות מרובות

‫Kafka Connect משתמש במחברים כדי להעביר נתונים בין מערכות. צד אחד של המחבר תמיד מתקשר עם אשכול של שירות מנוהל ל-Apache Kafka. אפשר להריץ כמה מחברים באותו אשכול של Kafka Connect. כל מחבר פועל כמקור (ששולף נתונים ממערכת) או כיעד (שדוחף נתונים למערכת).

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

לדוגמה, אם קלאסטר של השירות המנוהל ל-Apache Kafka נמצא ב-region-a, קלאסטר Connect נמצא ב-region-b ואתם משתמשים בחיבור ל-Cloud Storage לקטגוריה ב-region-c, תחויבו על שני צעדים ברשת (region-a ל-region-b ואז region-b ל-region-c, או הפוך, בהתאם לכיוון המחבר).

כשמתכננים את המיקום של אשכול Connect, חשוב לשקול היטב את כל האזורים הרלוונטיים כדי לייעל את זמן האחזור והעלות.

הגדרת הקיבולת

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

  • מעבדים וירטואליים: מספר המעבדים הווירטואליים שהוקצו לאשכול Connect. הערך המינימלי הוא 3 vCPU.

  • זיכרון: כמות הזיכרון שמוקצית לכל vCPU. צריך להקצות בין 1GB ל-8GB לכל vCPU. אחרי שיוצרים את האשכול, אפשר להגדיל או להקטין את כמות הזיכרון במסגרת המגבלות האלה.

    לדוגמה, אם יוצרים אשכול עם 6 מעבדי CPU וירטואליים, הזיכרון המינימלי שאפשר להקצות לאשכול הוא 6GB‎‏ (1GB‎ לכל מעבד CPU וירטואלי), והזיכרון המקסימלי הוא 48GB‎‏ (8GB‎ לכל מעבד CPU וירטואלי).

ל-vCPU ולהקצאת הזיכרון שהוקצו לכל worker באשכול Connect יש השפעה משמעותית על הביצועים, הקיבולת והעלות של האשכול. בהמשך מפורט איך ליבות CPU וזיכרון וירטואלי משפיעים על אשכול Connect.

מספר ה-vCPU

  • ‫Kafka Connect מחלק את העבודה של מחבר למשימות. כל משימה יכולה לעבד נתונים במקביל. יותר מעבדים וירטואליים מאפשרים להריץ יותר משימות בו-זמנית, וכך להגדיל את קצב העברת הנתונים.

  • הוספה של עוד מעבדי vCPU מגדילה את העלויות של אשכול Connect.

זיכרון

  • ‫Kafka Connect משתמש בזיכרון כדי לשמור נתונים במאגר זמני בזמן שהם עוברים בין מחברים לבין שירות מנוהל ל-Apache Kafka. זיכרון גדול יותר מאפשר שימוש במאגרי נתונים גדולים יותר. זיכרון גדול יכול לשפר את קצב העברת הנתונים, במיוחד בזרמי נתונים בכמות גדולה. למחברים שמתמודדים עם הודעות או רשומות גדולות מאוד נדרשת כמות זיכרון מספקת כדי לעבד אותן בלי להיתקל בחריגים OutOfMemoryError.

  • ככל שהזיכרון גדול יותר, העלות של אשכול Connect גבוהה יותר.

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

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

רשת משנה של Worker (ראשי)

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

אלה כמה דרישות להגדרת רשת המשנה של העובד:

  • חובה לציין את רשת המשנה של העובד.

  • רשת המשנה צריכה להיות באותו אזור שבו נמצא אשכול Connect.

  • רשת המשנה צריכה להיות באותו VPC ראשי כמו אחת מרשימת רשתות המשנה המחוברות של אשכול Kafka הראשי.

  • טווח ה-CIDR של רשת המשנה צריך להיות בגודל מינימלי של ‎ /22 (1,024 כתובות).

כתובות ה-IP מוקצות לעובדים באשכול בתת-הרשת של העובדים, באמצעות ממשק Private Service Connect. העובדים יכולים להגיע לכל יעד ברשת שאפשר לגשת אליו מרשת ה-VPC של רשת המשנה, בכפוף לדרישות הבאות:

  • נקודת הקצה לא יכולה להיות בטווח ה-CIDR‏ 172.16.0.0/14. הטווח הזה שמור לשימוש פנימי של שירות מנוהל ל-Apache Kafka Connect.
  • כללי חומת האש צריכים לאפשר את התעבורה. איך מגדירים אבטחה לקבצים שמצורפים לרשת
  • כדי להעביר תעבורת נתונים באינטרנט, צריך להגדיר Cloud NAT. לדוגמה, נדרש Cloud NAT כדי שמחבר MirrorMaker יוכל לשכפל נתונים מאשכול Kafka שאפשר לגשת אליו דרך האינטרנט.
  • כדי לגשת לנקודות קצה של Private Service Connect שנמצאות ב-VPC אחר מרשת המשנה של העובד, צריך לוודא שאתם משתמשים בהגדרת צרכן נתמכת (לדוגמה, NCC). מידע נוסף זמין במאמר גישה לשירותים שפורסמו דרך נקודות קצה.

דומייני DNS שניתן לפתור

דומיינים של DNS שאפשר לפתור, שנקראים גם שמות דומיין של DNS, מאפשרים להפוך כתובות DNS ברשת ה-VPC של הצרכן לזמינות ב-VPC של הדייר. כך אפשר לפתור את שמות ה-DNS של אשכול Connect לכתובות IP, ולתקשר עם שירותים אחרים, כולל אשכולות Kafka אחרים עבור מחברי MirrorMaker.

בדומיינים של DNS שאפשר לפתור, אפשר לבחור באשכול של שירות מנוהל ל-Apache Kafka. אין צורך להגדיר את שם הדומיין של ה-DNS עבור האשכול הראשי של השירות המנוהל ל-Apache Kafka, כי כתובת ה-bootstrap שלו נכללת אוטומטית ברשימת הדומיינים של ה-DNS שניתנים לפתרון.

עם זאת, אפשר גם לציין דומיין DNS באופן ידני, וזה נדרש אם בוחרים באשכול Kafka חיצוני. דומיין ה-DNS של האשכול הראשי של השירות המנוהל ל-Apache Kafka נכלל באופן אוטומטי. עדיין צריך להגדיר דומייני DNS עבור אשכולות Kafka אחרים.

משאבים של Secret Manager

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

כדי להשתמש בסודות של Secret Manager עם Kafka Connect, צריך:

  1. מקצים את התפקיד Secret Manager Secret Accessor ‏(roles/secretmanager.secretAccessor) לחשבון השירות של שירות Kafka המנוהל. התפקיד הזה מאפשר לאשכול Connect לגשת לסודות.

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

  3. כשיוצרים או מעדכנים את אשכול Connect, צריך לציין את הסודות שיש לאשכול גישה אליהם. אפשר לציין עד 32 ערכי Secret לכל אשכול Connect.

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

  • כדי להפנות לנתיב של קובץ סודי, צריך להשתמש בפורמט הבא:

    /var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION

    לדוגמה: ssl.truststore.location=/var/secrets/project1-truststore-1

  • כדי להשתמש בערך של סוד כערך הגדרה (לדוגמה, סיסמה), צריך להשתמש בפורמט הבא:

    ${directory:/var/secrets:PROJECT_NAME-SECRET_NAME-SECRET_VERSION}

    לדוגמה: password=${directory:/var/secrets:project1-database_password-3}

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

  • ‫PROJECT_NAME: השם של Google Cloud הפרויקט.
  • ‫SECRET_NAME: השם של הסוד.
  • ‫SECRET_VERSION: גרסת הסוד.

תוויות

תוויות הן צמדי מפתח/ערך שעוזרים לכם לארגן ולזהות את המשאבים. הם עוזרים לכם לארגן את אשכולי Connect. אפשר לצרף תווית לכל אשכול Connect כדי לסנן את המשאבים לפי התוויות. דוגמאות לתוויות: environment:prod, application:web-app.

יצירת אשכול Connect

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

יצירת אשכול Connect נמשכת 20 עד 30 דקות.

המסוף

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

    מעבר אל Connect Clusters

  2. לוחצים על Create.

  3. בשדה Connect cluster name (שם אשכול Connect), מזינים שם לאשכול Connect. מידע נוסף זמין במאמר הנחיות למתן שמות למשאבים בשירות מנוהל ל-Apache Kafka.

  4. ברשימה Primary Kafka cluster, בוחרים באשכול של שירות מנוהל ל-Apache Kafka. מידע נוסף זמין במאמר בנושא אשכול Kafka ראשי.

  5. ברשימה Region, בוחרים מיקום לאשכול Connect. מידע נוסף על בחירת מיקום זמין במאמר אשכול Kafka ראשי.

  6. בקטע Capacity configuration (הגדרת קיבולת), מזינים ערכים בשדות הבאים או משאירים את ערכי ברירת המחדל.

    • בשדה vCPUs, מזינים את מספר המעבדים הווירטואליים לאשכול.

    • בשדה זיכרון, מזינים את נפח הזיכרון לכל מעבד, ב-GiB. הערך לא יכול להיות גבוה מ-8GiB לכל CPU.

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

  7. בקטע Network configuration (הגדרת רשת), בוחרים רשת VPC מהרשימה Network (רשת) או משאירים את ערך ברירת המחדל. הרשימה הזו מתמלאת כשבוחרים את אשכול Kafka הראשי.

  8. בקטע Worker subnet, בוחרים רשת משנה מהרשימה Subnet או משאירים את ערך ברירת המחדל. מידע נוסף זמין במאמר בנושא רשת משנה של Worker. השדה Subnet URI path (נתיב URI של רשת משנה) מתמלא אוטומטית כשבוחרים את רשת המשנה.

  9. אופציונלי: מוסיפים דומיין DNS שניתן לפתור. דומיין ה-DNS של אשכול Kafka הראשי מתווסף באופן אוטומטי כדומיין DNS שניתן לפתור. כדי לציין דומיינים נוספים של DNS:

    1. מרחיבים את הקטע דומיינים של DNS שאפשר לפתור.

    2. לוחצים על הוספת דומיין DNS.

    3. כדי להוסיף את דומיין ה-DNS של אשכול קיים של שירות מנוהל ל-Apache Kafka, בוחרים את האשכול מהרשימה Kafka cluster (אשכול Kafka). אחרת, מזינים את דומיין ה-DNS בשדה דומיין ה-DNS.

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

  10. אופציונלי: כדי להוסיף משאבים של Secret Manager, מבצעים את הפעולות הבאות:

    1. מרחיבים את הקטע Secret manager resources.

    2. לוחצים על הוספת מקור סודי.

    3. ברשימה Secret, בוחרים סוד.

    4. ברשימה Secret version (גרסת הסוד), בוחרים גרסה של הסוד.

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

  11. אופציונלי: מוסיפים תוויות כדי לארגן את הפרויקט. כדי להוסיף תווית:

    1. מרחיבים את הקטע Labels.

    2. לוחצים על הוספת תווית.

    3. בשדה Key (מפתח), מזינים את המפתח של התווית.

    4. בשדה ערך, מזינים את הערך של התווית.

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

gcloud

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

    הפעלת Cloud Shell

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

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

    gcloud managed-kafka connect-clusters create CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU \
    --memory=MEMORY \
    --primary-subnet=WORKER_SUBNET \
    --kafka-cluster=KAFKA_CLUSTER \
    [--project=PROJECT_ID] \
    [--secret=SECRET] \
    [--dns-name=DNS_DOMAIN_NAME] \
    [--config-file=CONFIG_FILE] \
    [--labels=LABELS]
    [--async]

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

    • ‫CONNECT_CLUSTER_ID: המזהה או השם של אשכול Connect. הנחיות למתן שם לאשכול Connect מופיעות במאמר הנחיות למתן שם למשאב של שירות מנוהל ל-Apache Kafka. אי אפשר לשנות את השם של אשכול Connect.

    • LOCATION: המיקום שבו יוצרים את אשכול Connect. האזור חייב להיות אזור נתמך Google Cloud. אי אפשר לשנות את המיקום של אשכול Connect אחרי שיוצרים אותו. לרשימת המיקומים הזמינים, אפשר לעיין במאמר בנושא מיקומים של שירות מנוהל ל-Apache Kafka. מידע נוסף על המלצות למיקום זמין במאמר אשכול Kafka ראשי.

    • ‫CPU: מספר ליבות ה-vCPU של אשכול Connect. הערך המינימלי הוא 3 vCPUs. מידע נוסף על מספר ליבות ה-vCPU

    • ‫MEMORY: כמות הזיכרון של אשכול Connect. צריך להשתמש ביחידות 'MB',‏ 'MiB',‏ 'GB',‏ 'GiB',‏ 'TB' או 'TiB'. לדוגמה, ‎3GiB. צריך להקצות בין 1 ל-8 גיגה-בייט לכל CPU וירטואלי. מידע נוסף על זיכרון

    • ‫WORKER_SUBNET: תת-הרשת של העובד באשכול Connect.

      הפורמט של רשת המשנה הוא projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.

      רשת המשנה של העובד צריכה להיות באותו אזור כמו אשכול Connect.

    • ‫PROJECT_ID: (אופציונלי) מזההGoogle Cloud הפרויקט. אם לא מציינים פרויקט, המערכת משתמשת בפרויקט הנוכחי.

    • ‫KAFKA_CLUSTER: המזהה או השם המלא של האשכול הראשי של השירות המנוהל ל-Apache Kafka שמשויך לאשכול Connect. מידע נוסף זמין במאמר בנושא Kafka cluster. הפורמט של אשכול Kafka הוא projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID.

      אחרי שיוצרים את אשכול Connect, אי אפשר לעדכן אותו לאשכול Kafka אחר.

    • ‫SECRET: (אופציונלי) סודות לטעינה לעובדים. צריך לספק גרסאות מדויקות של סודות מ-Secret Manager, אין תמיכה בכינויים. אפשר לטעון עד 32 ערכי Secret לאותו אשכול. פורמט: projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID

    • ‫DNS_DOMAIN_NAME: (אופציונלי) שמות דומיין של DNS מתת-הרשת שרוצים להפוך לגלויים ל-Connect Cluster. ל-Connect cluster יש גישה למשאבים באמצעות שמות דומיין, במקום להסתמך על כתובות IP. כדאי לעיין במאמר בנושא DNS peering.

    • ‫LABELS: (אופציונלי) תוויות לשיוך לאשכול. מידע נוסף על הפורמט של תוויות זמין במאמר בנושא תוויות. רשימה של צמדי מפתח=ערך של תוויות להוספה. המפתחות צריכים להתחיל באות קטנה, ויכולים להכיל רק מקפים (-), קווים תחתונים (_), אותיות קטנות וספרות. הערכים יכולים להכיל רק מקפים (-), קווים תחתונים (_), אותיות קטנות וספרות.

    • ‫CONFIG_FILE: (אופציונלי) הנתיב לקובץ JSON או YAML שמכיל את ההגדרה שמוגדרת במקום ברירת המחדל של האשכול או המחבר. הקובץ הזה תומך גם ב-JSON או ב-YAML מוטבעים.

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

    מקבלים תגובה שדומה לזו:

    Create request issued for: [sample-connectcluster]
Check operation [projects/test-project/locations/us-east1/operations/operation-1753590328249-63ae19098cc06-64300a0a-06512d02] for status.

    מאחסנים את OPERATION_ID כדי לעקוב אחרי ההתקדמות. לדוגמה, הערך כאן הוא operation-1753590328249-63ae19098cc06-64300a0a-06512d02.

Terraform

אפשר להשתמש במשאב של Terraform כדי ליצור אשכול Connect.

resource "google_managed_kafka_connect_cluster" "default" {
  provider           = google-beta
  project            = data.google_project.default.project_id
  connect_cluster_id = "my-connect-cluster-id"
  location           = "us-central1"
  kafka_cluster      = google_managed_kafka_cluster.default.id
  capacity_config {
    vcpu_count   = 12
    memory_bytes = 12884901888 # 12 GiB
  }
  gcp_config {
    access_config {
      network_configs {
        primary_subnet = google_compute_subnetwork.default.id
      }
    }
  }
}

כדי ללמוד איך להחיל הגדרות ב-Terraform או להסיר אותן, ראו פקודות בסיסיות ב-Terraform.

המשך

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Go בקטע התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של ה-API של Go עבור שירות מנוהל ל-Apache Kafka.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, צריך להגדיר את Application Default Credentials‏(ADC). מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func createConnectCluster(w io.Writer, projectID, region, clusterID, kafkaCluster string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-connect-cluster"
	// kafkaCluster := "projects/my-project-id/locations/us-central1/clusters/my-kafka-cluster"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	locationPath := fmt.Sprintf("projects/%s/locations/%s", projectID, region)
	clusterPath := fmt.Sprintf("%s/connectClusters/%s", locationPath, clusterID)

	// Capacity configuration with 12 vCPU and 12 GiB RAM
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   12,
		MemoryBytes: 12884901888, // 12 GiB in bytes
	}

	// Optionally, you can also specify accessible subnets and resolvable DNS
	// domains as part of your network configuration. For example:
	// networkConfigs := []*managedkafkapb.ConnectNetworkConfig{
	// 	{
	// 		PrimarySubnet:      primarySubnet,
	// 		AdditionalSubnets:  []string{"subnet-1", "subnet-2"},
	// 		DnsDomainNames:     []string{"domain-1", "domain-2"},
	// 	},
	// }

	connectCluster := &managedkafkapb.ConnectCluster{
		Name:           clusterPath,
		KafkaCluster:   kafkaCluster,
		CapacityConfig: capacityConfig,
	}

	req := &managedkafkapb.CreateConnectClusterRequest{
		Parent:           locationPath,
		ConnectClusterId: clusterID,
		ConnectCluster:   connectCluster,
	}
	op, err := client.CreateConnectCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnectCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 5-15 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created connect cluster: %s\n", resp.Name)
	return nil
}

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במאמר התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של Managed Service for Apache Kafka Java API.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, מגדירים את ה-Application Default Credentials. מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.


import com.google.api.gax.longrunning.OperationFuture;
import com.google.api.gax.longrunning.OperationSnapshot;
import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.RetryingFuture;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.ConnectAccessConfig;
import com.google.cloud.managedkafka.v1.ConnectCluster;
import com.google.cloud.managedkafka.v1.ConnectGcpConfig;
import com.google.cloud.managedkafka.v1.ConnectNetworkConfig;
import com.google.cloud.managedkafka.v1.CreateConnectClusterRequest;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateConnectCluster {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String clusterId = "my-connect-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    String kafkaCluster = "my-kafka-cluster"; // The Kafka cluster to connect to
    int cpu = 12;
    long memoryBytes = 12884901888L; // 12 GiB
    createConnectCluster(projectId, region, clusterId, subnet, kafkaCluster, cpu, memoryBytes);
  }

  public static void createConnectCluster(
      String projectId,
      String region,
      String clusterId,
      String subnet,
      String kafkaCluster,
      int cpu,
      long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setVcpuCount(cpu)
        .setMemoryBytes(memoryBytes).build();
    ConnectNetworkConfig networkConfig = ConnectNetworkConfig.newBuilder()
        .setPrimarySubnet(subnet)
        .build();
    // Optionally, you can also specify additional accessible subnets and resolvable
    // DNS domains as part of your network configuration. For example:
    // .addAllAdditionalSubnets(List.of("subnet-1", "subnet-2"))
    // .addAllDnsDomainNames(List.of("dns-1", "dns-2"))
    ConnectGcpConfig gcpConfig = ConnectGcpConfig.newBuilder()
        .setAccessConfig(ConnectAccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
        .build();
    ConnectCluster connectCluster = ConnectCluster.newBuilder()
        .setCapacityConfig(capacityConfig)
        .setGcpConfig(gcpConfig)
        .setKafkaCluster(kafkaCluster)
        .build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaConnectSettings.Builder settingsBuilder = ManagedKafkaConnectSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.createConnectClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient
        .create(settingsBuilder.build())) {
      CreateConnectClusterRequest request = CreateConnectClusterRequest.newBuilder()
          .setParent(LocationName.of(projectId, region).toString())
          .setConnectClusterId(clusterId)
          .setConnectCluster(connectCluster)
          .build();

      // The duration of this operation can vary considerably, typically taking
      // between 10-30 minutes.
      OperationFuture<ConnectCluster, OperationMetadata> future = managedKafkaConnectClient
          .createConnectClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf(
          "Connect cluster creation started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(), operation.isDone(), future.getMetadata().get().toString());

      while (!future.isDone()) {
        // The pollingFuture gives us the most recent status of the operation
        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
            currentOp.getName(),
            currentOp.isDone());
      }

      // NOTE: future.get() blocks completion until the operation is complete (isDone
      // = True)
      ConnectCluster response = future.get();
      System.out.printf("Created connect cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaConnectClient.createConnectCluster got err: %s\n", 
          e.getMessage());
      throw e;
    }
  }
}

Python

לפני שמנסים את הדוגמה הזו, פועלים לפי הוראות ההגדרה של Python במאמר התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של ה-API בשפת Python של שירות מנוהל ל-Apache Kafka.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, מגדירים את ה-Application Default Credentials. מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1
from google.cloud.managedkafka_v1.services.managed_kafka_connect import ManagedKafkaConnectClient
from google.cloud.managedkafka_v1.types import ConnectCluster, CreateConnectClusterRequest, ConnectNetworkConfig

# TODO(developer): Update with your values.
# project_id = "my-project-id"
# region = "us-central1"
# connect_cluster_id = "my-connect-cluster"
# kafka_cluster_id = "my-kafka-cluster"
# primary_subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 12
# memory_bytes = 12884901888  # 12 GiB

connect_client = ManagedKafkaConnectClient()
kafka_client = managedkafka_v1.ManagedKafkaClient()

parent = connect_client.common_location_path(project_id, region)
kafka_cluster_path = kafka_client.cluster_path(project_id, region, kafka_cluster_id)

connect_cluster = ConnectCluster()
connect_cluster.name = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)
connect_cluster.kafka_cluster = kafka_cluster_path
connect_cluster.capacity_config.vcpu_count = cpu
connect_cluster.capacity_config.memory_bytes = memory_bytes
connect_cluster.gcp_config.access_config.network_configs = [ConnectNetworkConfig(primary_subnet=primary_subnet)]
# Optionally, you can also specify accessible subnets and resolvable DNS domains as part of your network configuration.
# For example:
# connect_cluster.gcp_config.access_config.network_configs = [
#     ConnectNetworkConfig(
#         primary_subnet=primary_subnet,
#         additional_subnets=additional_subnets,
#         dns_domain_names=dns_domain_names,
#     )
# ]

request = CreateConnectClusterRequest(
    parent=parent,
    connect_cluster_id=connect_cluster_id,
    connect_cluster=connect_cluster,
)

try:
    operation = connect_client.create_connect_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # Creating a Connect cluster can take 10-40 minutes.
    response = operation.result(timeout=3000)
    print("Created Connect cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

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

אפשר להריץ את הפקודה הבאה רק אם הפעלתם את ה-CLI של gcloud כדי ליצור את אשכול Connect.

  • יצירת אשכול Connect נמשכת בדרך כלל 20-30 דקות. כדי לעקוב אחרי התקדמות היצירה של האשכול, הפקודה gcloud managed-kafka connect-clusters create משתמשת בפעולה ארוכת טווח (LRO), שאפשר לעקוב אחריה באמצעות הפקודה הבאה:

    gcloud managed-kafka operations describe OPERATION_ID \
    --location=LOCATION

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

    • ‫OPERATION_ID עם הערך של מזהה הפעולה מהקטע הקודם.
    • ‫LOCATION עם ערך המיקום מהקטע הקודם.

מה השלב הבא?

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