קישור של Pub/Sub ל-Apache Kafka

במאמר הזה מוסבר איך לשלב בין Apache Kafka ו-Pub/Sub באמצעות Pub/Sub Group Kafka Connector.

מידע על Pub/Sub Group Kafka Connector

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

באמצעות Pub/Sub Group Kafka Connector אפשר לשלב בין שתי המערכות האלה. המחברים הבאים ארוזים בקובץ ה-JAR של המחבר:

  • מחבר ה-sink קורא רשומות מנושא אחד או יותר ב-Kafka ומפרסם אותן ב-Pub/Sub.
  • המחבר של המקור קורא הודעות מנושא Pub/Sub ומפרסם אותן ב-Kafka.

הנה כמה תרחישים שבהם אפשר להשתמש ב-Pub/Sub Group Kafka Connector:

  • אתם מעבירים ארכיטקטורה מבוססת-Kafka אל Cloud de Confiance by S3NS.
  • יש לכם מערכת frontend שמאחסנת אירועים ב-Kafka מחוץ ל-Cloud de Confiance by S3NS, אבל אתם גם משתמשים ב- Cloud de Confiance by S3NS כדי להפעיל חלק משירותי ה-backend שלכם, שצריכים לקבל את האירועים ב-Kafka.
  • אתם אוספים יומנים מפתרון Kafka מקומי ושולחים אותם אל Cloud de Confiance by S3NS לניתוח נתונים.
  • יש לכם מערכת frontend שמשתמשת ב- Cloud de Confiance by S3NS, אבל אתם גם מאחסנים נתונים באופן מקומי באמצעות Kafka.

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

ההנחה במאמר הזה היא שאתם מכירים את Kafka ואת Pub/Sub. לפני שקוראים את המאמר הזה, מומלץ לעבור על אחד ממדריכי ההתחלה המהירה של Pub/Sub.

המחבר של Pub/Sub לא תומך בשילוב בין Cloud de Confiance IAM לבין רשימות בקרת גישה (ACL) של Kafka Connect.

תחילת העבודה עם המחבר

בקטע הזה מוסבר איך לבצע את המשימות הבאות:

  1. מגדירים את Pub/Sub Group Kafka Connector.
  2. שליחת אירועים מ-Kafka ל-Pub/Sub.
  3. שליחת הודעות מ-Pub/Sub ל-Kafka.

דרישות מוקדמות

התקנת Kafka

פועלים לפי המדריך לתחילת העבודה עם Apache Kafka כדי להתקין Kafka עם צומת יחיד במחשב המקומי. מבצעים את השלבים הבאים במדריך למתחילים:

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

אמת

כדי לשלוח ולקבל הודעות Pub/Sub, מחבר Pub/Sub Group Kafka צריך לעבור אימות ב-Pub/Sub. כדי להגדיר אימות, מבצעים את השלבים הבאים:

  1. התקינו את ה-CLI של Google Cloud.

  2. הגדירו שה-CLI של gcloud ישתמש בזהות המאוחדת שלכם.

    איך נכנסים ל-CLI של gcloud באמצעות הזהות המאוחדת?

  3. כדי לאתחל את ה-CLI של gcloud, הריצו את הפקודה הבאה: 

    gcloud init

  4. יוצרים או בוחרים Cloud de Confiance פרויקט.

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

    • Select a project: כדי לבחור פרויקט לא צריך תפקיד IAM ספציפי – אפשר לבחור כל פרויקט שקיבלתם בו תפקיד.
    • יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד Project Creator (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים?

    • יוצרים Cloud de Confiance פרויקט:

      gcloud projects create PROJECT_ID

      מחליפים את PROJECT_ID בשם של פרויקט Cloud de Confiance שיוצרים.

    • בוחרים את הפרויקט שיצרתם ב- Cloud de Confiance :

      gcloud config set project PROJECT_ID

      מחליפים את PROJECT_ID בשם הפרויקט ב- Cloud de Confiance .

  5. צרו פרטי כניסה לאימות מקומי בחשבון המשתמש: 

    gcloud auth application-default login

    אם מוחזרת שגיאת אימות, צריך לוודא ש נכנסתם ל-CLI של gcloud באמצעות הזהות המאוחדת שלכם.

  6. נותנים תפקידים לחשבון המשתמש. מריצים את הפקודה הבאה לכל אחד מהתפקידים הבאים ב-IAM: roles/pubsub.admin 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

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

הורדת קובץ ה-JAR של המחבר

מורידים את קובץ ה-JAR של המחבר למחשב המקומי. מידע נוסף זמין בקטע Acquire the connector בקובץ ה-README ב-GitHub.

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

  1. משכפלים או מורידים את מאגר GitHub של המחבר.

    git clone https://github.com/googleapis/java-pubsub-group-kafka-connector.git
cd java-pubsub-group-kafka-connector

  2. מעתיקים את התוכן של הספרייה config לספריית המשנה config של ההתקנה של Kafka.

    cp config/* [path to Kafka installation]/config/

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

עדכון ההגדרה של Kafka Connect

  1. עוברים לספרייה שמכילה את הקובץ הבינארי של Kafka Connect שהורדתם.
  2. בספרייה הבינארית של Kafka Connect, פותחים את הקובץ config/connect-standalone.properties בכלי לעריכת טקסט.
  3. אם השורה plugin.path property מופיעה כהערה, צריך לבטל את ההערה.

  4. מעדכנים את plugin.path property כך שיכלול את הנתיב לקובץ ה-JAR של המחבר.

    דוגמה:

    plugin.path=/home/PubSubKafkaConnector/pubsub-group-kafka-connector-1.0.0.jar

  5. מגדירים את המאפיין offset.storage.file.filename לשם של קובץ מקומי. במצב עצמאי, Kafka משתמש בקובץ הזה כדי לאחסן נתוני אופסט.

    דוגמה:

    offset.storage.file.filename=/tmp/connect.offsets

העברת אירועים מ-Kafka ל-Pub/Sub

בקטע הזה מוסבר איך להפעיל את מחבר היעד, לפרסם אירועים ב-Kafka ואז לקרוא את ההודעות שהועברו מ-Pub/Sub.

  1. משתמשים ב-Google Cloud CLI כדי ליצור נושא Pub/Sub עם מינוי.

    gcloud pubsub topics create PUBSUB_TOPIC
gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC

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

    • ‫PUBSUB_TOPIC: השם של נושא ב-Pub/Sub שאליו יתקבלו ההודעות מ-Kafka.
    • ‫PUBSUB_SUBSCRIPTION: השם של מינוי Pub/Sub לנושא.

  2. פותחים את הקובץ /config/cps-sink-connector.properties בכלי לעריכת טקסט. מוסיפים ערכים למאפיינים הבאים, שמסומנים ב-"TODO" בהערות:

    topics=KAFKA_TOPICS
cps.project=PROJECT_ID
cps.topic=PUBSUB_TOPIC

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

    • ‫KAFKA_TOPICS: רשימה מופרדת בפסיקים של נושאי Kafka שמהם רוצים לקרוא.
    • ‫PROJECT_ID: הפרויקט Cloud de Confiance by S3NS שכולל את נושא ה-Pub/Sub.
    • ‫PUBSUB_TOPIC: נושא Pub/Sub לקבלת ההודעות מ-Kafka.

  3. מריצים את הפקודה הבאה מהספרייה של Kafka:

    bin/connect-standalone.sh \
  config/connect-standalone.properties \
  config/cps-sink-connector.properties

  4. פועלים לפי השלבים בהפעלה מהירה של Apache Kafka כדי לכתוב כמה אירועים לנושא Kafka.

  5. משתמשים ב-CLI של gcloud כדי לקרוא את האירועים מ-Pub/Sub.

    gcloud pubsub subscriptions pull PUBSUB_SUBSCRIPTION --auto-ack

העברת הודעות מ-Pub/Sub ל-Kafka

בקטע הזה מוסבר איך להפעיל את מחבר המקור, לפרסם הודעות ב-Pub/Sub ולקרוא את ההודעות שהועברו מ-Kafka.

  1. משתמשים ב-CLI של gcloud כדי ליצור נושא Pub/Sub עם מינוי.

    gcloud pubsub topics create PUBSUB_TOPIC
gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC

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

    • ‫PUBSUB_TOPIC: השם של נושא Pub/Sub.
    • ‫PUBSUB_SUBSCRIPTION: השם של מינוי ל-Pub/Sub.

  2. פותחים את הקובץ בשם /config/cps-source-connector.properties בכלי לעריכת טקסט. מוסיפים ערכים למאפיינים הבאים, שמסומנים ב-"TODO" בהערות:

    kafka.topic=KAFKA_TOPIC
cps.project=PROJECT_ID
cps.subscription=PUBSUB_SUBSCRIPTION

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

    • ‫KAFKA_TOPIC: נושא Kafka לקבלת הודעות Pub/Sub.
    • ‫PROJECT_ID: הפרויקט Cloud de Confiance by S3NS שכולל את נושא ה-Pub/Sub.
    • ‫PUBSUB_TOPIC: נושא ה-Pub/Sub.

  3. מריצים את הפקודה הבאה מהספרייה של Kafka:

    bin/connect-standalone.sh \
  config/connect-standalone.properties \
  config/cps-source-connector.properties

  4. משתמשים ב-CLI של gcloud כדי לפרסם הודעה ב-Pub/Sub.

    gcloud pubsub topics publish PUBSUB_TOPIC --message="message 1"

  5. קוראים את ההודעה מ-Kafka. כדי לקרוא את ההודעות מנושא Kafka, פועלים לפי השלבים שבמדריך למתחילים של Apache Kafka.

המרה מהודעת טקסט

רשומה ב-Kafka מכילה מפתח וערך, שהם מערכי בייטים באורך משתנה. רשומת Kafka יכולה גם לכלול כותרות, שהן צמדי מפתח/ערך. הודעת Pub/Sub כוללת שני חלקים עיקריים: גוף ההודעה ואפס או יותר מאפיינים של צמדי מפתח/ערך.

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

  • ‫key.converter: הממיר שמשמש לסריאליזציה של מפתחות הרשומות.
  • ‫value.converter: הממיר שמשמש לסריאליזציה של ערכי הרשומות.

גוף ההודעה ב-Pub/Sub הוא אובייקט ByteString, ולכן ההמרה היעילה ביותר היא להעתיק את המטען הייעודי (payload) ישירות. לכן, מומלץ להשתמש בממיר שמפיק סוגי נתונים פרימיטיביים (מספר שלם, מספר עשרוני, מחרוזת או סכימת בייטים) איפה שאפשר, כדי למנוע את הסרת הסריאליזציה והסריאליזציה מחדש של אותו גוף הודעה.

המרת נתונים מ-Kafka ל-Pub/Sub

מחבר היעד ממיר רשומות של Kafka להודעות Pub/Sub באופן הבא:

  • מפתח הרשומה של Kafka מאוחסן כמאפיין בשם "key" בהודעה ב-Pub/Sub.
  • כברירת מחדל, המחבר משמיט את כל הכותרות ברשומת Kafka. עם זאת, אם מגדירים את אפשרות ההגדרה headers.publish לערך true, המחבר כותב את הכותרות כמאפיינים של Pub/Sub. המחבר מדלג על כותרות שחורגות מהמגבלות של Pub/Sub על מאפייני הודעה.
  • בסכימות של מספרים שלמים, מספרים עשרוניים, מחרוזות ובייטים, המחבר מעביר את הבייטים של ערך הרשומה ב-Kafka ישירות לגוף ההודעה ב-Pub/Sub.
  • בסכימות של מבנים, המחבר כותב כל שדה כמאפיין של הודעת Pub/Sub. לדוגמה, אם השדה הוא { "id"=123 }, ההודעה שמתקבלת ב-Pub/Sub כוללת מאפיין "id"="123". הערך של השדה תמיד מומר למחרוזת. סוגי מפות ומבנים לא נתמכים כסוגי שדות בתוך מבנה.
  • בסכימות של מפות, המחבר כותב כל צמד מפתח/ערך כמאפיין של הודעת Pub/Sub. לדוגמה, אם המיפוי הוא {"alice"=1,"bob"=2}, להודעת Pub/Sub שמתקבלת יש שני מאפיינים, "alice"="1" ו-"bob"="2". המפתחות והערכים מומרים למחרוזות.

לסכימות של struct ומפות יש כמה התנהגויות נוספות:

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

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

המרת נתונים מ-Pub/Sub ל-Kafka

מחבר המקור ממיר הודעות Pub/Sub לרשומות Kafka באופן הבא:

  • מפתח רשומה של Kafka: כברירת מחדל, המפתח מוגדר ל-null. אפשר גם לציין מאפיין של הודעת Pub/Sub שישמש כמפתח, על ידי הגדרת אפשרות ההגדרה kafka.key.attribute. במקרה כזה, המחבר מחפש מאפיין עם השם הזה ומגדיר את מפתח הרשומה לערך המאפיין. אם המאפיין שצוין לא קיים, מפתח הרשומה מוגדר ל-null.

  • ערך הרשומה ב-Kafka. המחבר כותב את ערך הרשומה באופן הבא:

    • אם להודעת Pub/Sub אין מאפיינים מותאמים אישית, המחבר כותב את גוף ההודעה של Pub/Sub ישירות לערך הרשומה של Kafka כסוג byte[], באמצעות הממיר שצוין על ידי value.converter.

    • אם הודעת Pub/Sub מכילה מאפיינים מותאמים אישית והערך של kafka.record.headers הוא false, המחבר כותב struct לערך הרשומה. המבנה מכיל שדה אחד לכל מאפיין, ושדה בשם "message" שהערך שלו הוא גוף ההודעה ב-Pub/Sub (מאוחסן כבייטים):

      {
  "message": "<Pub/Sub message body>",
  "<attribute-1>": "<value-1>",
  "<attribute-2>": "<value-2>",
  ....
}

      במקרה כזה, צריך להשתמש ב-value.converter שתואם לסכימות struct, כמו org.apache.kafka.connect.json.JsonConverter.

    • אם להודעת Pub/Sub יש מאפיינים מותאמים אישית והערך של kafka.record.headers הוא true, המחבר כותב את המאפיינים ככותרות של רשומת Kafka. הוא כותב את גוף ההודעה של Pub/Sub ישירות לערך הרשומה של Kafka כסוג byte[], באמצעות הממיר שצוין על ידי value.converter.

  • כותרות של רשומות Kafka. כברירת מחדל, הכותרות ריקות, אלא אם מגדירים את kafka.record.headers לערך true.

אפשרויות להגדרות אישיות

בנוסף להגדרות האישיות שסופקו על ידי Kafka Connect API, המחבר Pub/Sub Group Kafka Connector תומך בהגדרות האישיות של הכיור והמקור, כפי שמתואר במאמר הגדרות אישיות של מחבר Pub/Sub.

קבלת תמיכה

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

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