‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

יצירת מחבר של מקור Pub/Sub

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

תרחישי שימוש לדוגמה במחברי מקורות של Pub/Sub:

הטמעת נתונים בזמן אמת. שליחת נתונים משירותי ענן או מאפליקציות אחרות ל-Pub/Sub, ואז שכפול הנתונים ל-Kafka לעיבוד זרמי נתונים (stream processing).
ארכיטקטורות מבוססות-אירועים. הפעלת עיבוד מבוסס-Kafka מהודעות שפורסמו ב-Pub/Sub.

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

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

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

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

לפני שיוצרים מחבר של מקור Pub/Sub, צריך לוודא שיש לכם את הדברים הבאים:

נושא Pub/Sub עם מינוי.
אשכול של שירות מנוהל ל-Apache Kafka.
נושא Kafka בתוך אשכול Kafka.
אשכול Connect. כשיוצרים את אשכול Connect, מגדירים את אשכול השירות המנוהל ל-Apache Kafka כאשכול Kafka הראשי.

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

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

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

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

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

יצירת מחבר: managedkafka.connectors.create

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

צריך לתת הרשאות קריאה מ-Pub/Sub

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

Pub/Sub Subscriber (roles/pubsub.subscriber)
כלי הצפייה ב-Pub/Sub (roles/pubsub.viewer)

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

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

יצירת מחבר של מקור Pub/Sub

המסוף

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

מעבר אל Connect Clusters
לוחצים על Connect cluster (חיבור אשכול) במקום שבו רוצים ליצור את המחבר.
לוחצים על יצירת מחבר.
בשדה שם המחבר, מזינים מחרוזת.

הנחיות לשמות של מחברים זמינות במאמר הנחיות לשמות של משאבים בשירות מנוהל ל-Apache Kafka.
בשדה Connector plugin, בוחרים באפשרות Pub/Sub Source.
ברשימה Cloud Pub/Sub subscription, בוחרים מינוי ל-Pub/Sub. המחבר שולף הודעות מהמינוי הזה. המינוי מוצג כשם משאב מלא: projects/{project}/subscriptions/{subscription}.
ברשימה Kafka topic, בוחרים את נושא Kafka שבו נכתבות ההודעות.
אופציונלי: בתיבה Configurations, מוסיפים מאפייני הגדרה או עורכים את מאפייני ברירת המחדל. מידע נוסף מופיע במאמר הגדרת המחבר.
בוחרים באפשרות מדיניות הפעלה מחדש של משימות. מידע נוסף זמין במאמר בנושא מדיניות הפעלה מחדש של משימות.
לוחצים על יצירה.

gcloud

מריצים את הפקודה 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 או JSON.

דוגמה לקובץ תצורה:

connector.class: "com.google.pubsub.kafka.source.CloudPubSubSourceConnector"
cps.project: "PROJECT_ID"
cps.subscription: "PUBSUB_SUBSCRIPTION_ID"
kafka.topic: "KAFKA_TOPIC_ID"
value.converter: "org.apache.kafka.connect.converters.ByteArrayConverter"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
tasks.max: "3"

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

‫PROJECT_ID: המזהה של Google Cloud הפרויקט שבו נמצאת המינוי ל-Pub/Sub.
‫PUBSUB_SUBSCRIPTION_ID: המזהה של מינוי Pub/Sub שממנו שולפים נתונים.
‫KAFKA_TOPIC_ID: המזהה של נושא Kafka שבו נכתבים הנתונים.

צריך להגדיר את מאפייני התצורה cps.project, cps.subscription ו-kafka.topic. אפשרויות הגדרה נוספות מפורטות במאמר הגדרת המחבר.

Terraform

אתם יכולים להשתמש במשאב של Terraform כדי ליצור מחבר.

resource "google_managed_kafka_connector" "example-pubsub-source-connector" {
  project         = data.google_project.default.project_id
  connector_id    = "my-pubsub-source-connector"
  connect_cluster = google_managed_kafka_connect_cluster.default.connect_cluster_id
  location        = "us-central1"

  configs = {
    "connector.class"  = "com.google.pubsub.kafka.source.CloudPubSubSourceConnector"
    "name"             = "my-pubsub-source-connector"
    "tasks.max"        = "3"
    "kafka.topic"      = "GMK_TOPIC_ID"
    "cps.subscription" = "CPS_SUBSCRIPTION_ID"
    "cps.project"      = data.google_project.default.project_id
    "value.converter"  = "org.apache.kafka.connect.converters.ByteArrayConverter"
    "key.converter"    = "org.apache.kafka.connect.storage.StringConverter"
  }

  provider = google-beta
}

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

המשך

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

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

import (
	"context"
	"fmt"
	"io"

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

// createPubSubSourceConnector creates a Pub/Sub Source connector.
func createPubSubSourceConnector(w io.Writer, projectID, region, connectClusterID, connectorID, kafkaTopic, cpsSubscription, cpsProject, tasksMax, valueConverter, keyConverter string, opts ...option.ClientOption) error {
	// TODO(developer): Update with your config values. Here is a sample configuration:
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "CPS_SOURCE_CONNECTOR_ID"
	// kafkaTopic := "GMK_TOPIC_ID"
	// cpsSubscription := "CPS_SUBSCRIPTION_ID"
	// cpsProject := "GCP_PROJECT_ID"
	// tasksMax := "3"
	// valueConverter := "org.apache.kafka.connect.converters.ByteArrayConverter"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)

	// Pub/Sub Source sample connector configuration
	config := map[string]string{
		"connector.class":  "com.google.pubsub.kafka.source.CloudPubSubSourceConnector",
		"name":             connectorID,
		"tasks.max":        tasksMax,
		"kafka.topic":      kafkaTopic,
		"cps.subscription": cpsSubscription,
		"cps.project":      cpsProject,
		"value.converter":  valueConverter,
		"key.converter":    keyConverter,
	}

	connector := &managedkafkapb.Connector{
		Name:    fmt.Sprintf("%s/connectors/%s", parent, connectorID),
		Configs: config,
	}

	req := &managedkafkapb.CreateConnectorRequest{
		Parent:      parent,
		ConnectorId: connectorID,
		Connector:   connector,
	}

	resp, err := client.CreateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Created Pub/Sub source connector: %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.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreatePubSubSourceConnector {

  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 connectClusterId = "my-connect-cluster";
    String connectorId = "my-pubsub-source-connector";
    String pubsubProjectId = "my-pubsub-project-id";
    String subscriptionName = "my-subscription";
    String kafkaTopicName = "pubsub-topic";
    String connectorClass = "com.google.pubsub.kafka.source.CloudPubSubSourceConnector";
    String maxTasks = "3";
    String valueConverter = "org.apache.kafka.connect.converters.ByteArrayConverter";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createPubSubSourceConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        pubsubProjectId,
        subscriptionName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        valueConverter,
        keyConverter);
  }

  public static void createPubSubSourceConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String pubsubProjectId,
      String subscriptionName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String valueConverter,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("name", connectorId);
    configMap.put("tasks.max", maxTasks);
    configMap.put("kafka.topic", kafkaTopicName);
    configMap.put("cps.subscription", subscriptionName);
    configMap.put("cps.project", pubsubProjectId);
    configMap.put("value.converter", valueConverter);
    configMap.put("key.converter", keyConverter);

    Connector connector = Connector.newBuilder()
        .setName(
            ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
        .putAllConfigs(configMap)
        .build();

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      CreateConnectorRequest request = CreateConnectorRequest.newBuilder()
          .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
          .setConnectorId(connectorId)
          .setConnector(connector)
          .build();

      // This operation is being handled synchronously.
      Connector response = managedKafkaConnectClient.createConnector(request);
      System.out.printf("Created Pub/Sub Source connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

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

from google.api_core.exceptions import GoogleAPICallError
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector, CreateConnectorRequest

connect_client = ManagedKafkaConnectClient()
parent = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)

configs = {
    "connector.class": "com.google.pubsub.kafka.source.CloudPubSubSourceConnector",
    "name": connector_id,
    "tasks.max": tasks_max,
    "kafka.topic": kafka_topic,
    "cps.subscription": cps_subscription,
    "cps.project": cps_project,
    "value.converter": value_converter,
    "key.converter": key_converter,
}

connector = Connector()
connector.name = connector_id
connector.configs = configs

request = CreateConnectorRequest(
    parent=parent,
    connector_id=connector_id,
    connector=connector,
)

try:
    operation = connect_client.create_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Created Connector:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

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

הגדרת המחבר

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

רשימה מלאה של המאפיינים שספציפיים למחבר הזה זמינה במאמר הגדרות של מחבר מקור Pub/Sub.

מצב משיכה

מצב השליפה מציין איך המחבר מאחזר הודעות Pub/Sub. המצבים הבאים נתמכים:

מצב משיכה (ברירת מחדל). ההודעות נשלפות בקבוצות. כדי להפעיל את המצב הזה, מגדירים את cps.streamingPull.enabled=false. כדי להגדיר את גודל האצווה, מגדירים את המאפיין cps.maxBatchSize.

מידע נוסף על מצב משיכה זמין במאמר בנושא Pull API.
מצב משיכה בסטרימינג. האפשרות הזו מאפשרת להשיג את התפוקה המקסימלית ואת זמן האחזור הנמוך ביותר כשמאחזרים הודעות מ-Pub/Sub. כדי להפעיל את המצב הזה, צריך להגדיר את cps.streamingPull.enabled=true.

מידע נוסף על מצב משיכה של סטרימינג זמין במאמר בנושא StreamingPull API.

אם הפעלתם את משיכת הנתונים בסטרימינג, תוכלו לשפר את הביצועים על ידי הגדרת מאפייני התצורה הבאים:
- ‫cps.streamingPull.flowControlBytes: המספר המקסימלי של בייטים של הודעות בהמתנה לכל משימה.
- ‫cps.streamingPull.flowControlMessages: המספר המקסימלי של הודעות ממתינות לטיפול לכל משימה.
- ‫cps.streamingPull.maxAckExtensionMs: משך הזמן המרבי שבו המחבר מאריך את המועד האחרון להרשמה, באלפיות השנייה.
- ‫cps.streamingPull.maxMsPerAckExtension: משך הזמן המקסימלי שבו המחבר מאריך את המועד האחרון להרשמה לכל הארכה, באלפיות השנייה.
- ‫cps.streamingPull.parallelStreams: מספר הזרמים שמהם יתבצע שליפת ההודעות מהמינוי.

נקודת קצה של Pub/Sub

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

מחיצות Kafka

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

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

משתמשים שביצעו המרה

מגדירים את ממיר המפתחות ל-org.apache.kafka.connect.storage.StringConverter.

בהתאם להגדרות של כלי החיבור, מגדירים את ממיר הערכים לאחת מהאפשרויות הבאות:

org.apache.kafka.connect.converters.ByteArrayConverter
org.apache.kafka.connect.json.JsonConverter

מידע נוסף מופיע במאמר בנושא ערך הרשומה.

המרת הודעות

מחבר Pub/Sub Source ממיר הודעות Pub/Sub לרשומות Kafka. בקטעים הבאים מתואר תהליך ההמרה.

מפתח הרשומה

הערך של המפתח converter חייב להיות org.apache.kafka.connect.storage.StringConverter.

כברירת מחדל, מפתחות הרשומה הם null.
כדי להשתמש במאפיין של הודעת Pub/Sub כמפתח, מגדירים את kafka.key.attribute לשם המאפיין. לדוגמה: kafka.key.attribute=username.
כדי להשתמש במפתח ההזמנה של Pub/Sub כמפתח, מגדירים את kafka.key.attribute=orderingKey.

הקלטת כותרות

כברירת מחדל, כותרות הרשומות ריקות.

אם kafka.record.headers הוא true, מאפייני ההודעה של Pub/Sub נכתבים ככותרות של רשומות. כדי לכלול את מפתח ההזמנה, צריך להגדיר את הערך cps.makeOrderingKeyAttribute=true.

ערך הרשומה

ערכי הרשומות נכתבים כמערכי בייטים או כסוגי struct.

ערכי רשומות של מערך בייטים

אם הערך של kafka.record.headers הוא true, או אם בהודעת Pub/Sub אין מאפיינים מותאמים אישית, המחבר כותב את נתוני ההודעה כמערך בייטים. מגדירים את ממיר הערכים ל-org.apache.kafka.connect.converters.ByteArrayConverter.

ערכים של רשומות Struct

אם הערך של kafka.record.headers הוא false ולהודעה יש לפחות מאפיין מותאם אישית אחד, המחבר כותב את ערך הרשומה כ-struct. מגדירים את ממיר הערכים ל-org.apache.kafka.connect.json.JsonConverter.

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

‫message: נתוני ההודעה ב-Pub/Sub, בבייטים.
שדה לכל מאפיין של הודעת Pub/Sub. כדי לכלול את מפתח ההזמנה, מגדירים את cps.makeOrderingKeyAttribute=true.

לדוגמה, אם להודעה יש מאפיין username, ערך הרשומה ייראה כך:

{
  "message":"MESSAGE_DATA",
  "username":"Alice"
}

אם הערך של value.converter.schemas.enable הוא true, ה-struct כולל גם את מטען הייעודי (payload) וגם את הסכימה:

{
  "schema":
    {
      "type":"struct",
      "fields": [
        {
          "type":"bytes",
          "optional":false,
          "field":"message"
        },
        {
          "type":"string",
          "optional":false,
          "field":"username"
        }
      ],
      "optional":false
    },
    "payload": {
      "message":"MESSAGE_DATA",
      "username":"Alice"
    }
}

מה השלב הבא?

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

יצירת מחבר של מקור Pub/Sub קל לארגן דפים בעזרת אוספים אפשר לשמור ולסווג תוכן על סמך ההעדפות שלך.

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

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

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

צריך לתת הרשאות קריאה מ-Pub/Sub

יצירת מחבר של מקור Pub/Sub

המסוף

gcloud

Terraform

המשך

Java

Python

הגדרת המחבר

מצב משיכה

נקודת קצה של Pub/Sub

מחיצות Kafka

משתמשים שביצעו המרה

המרת הודעות

מפתח הרשומה

הקלטת כותרות

ערך הרשומה

ערכי רשומות של מערך בייטים

ערכים של רשומות Struct

מה השלב הבא?

יצירת מחבר של מקור Pub/Sub