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

יצירת מחבר BigQuery Sink

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

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

לפני שיוצרים מחבר BigQuery Sink, צריך לוודא שיש לכם:

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

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

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

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

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

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

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

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

מתן הרשאות כתיבה בטבלה ב-BigQuery

לסוכן השירות של Managed Kafka צריכה להיות הרשאה לכתוב הודעות בטבלה ב-BigQuery. מקצים לסוכן השירות את התפקיד "עריכת נתונים ב-BigQuery" (roles/bigquery.dataEditor) בפרויקט שמכיל את הטבלה.

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

פרמטרים מרכזיים להגדרה

ההמלצות הבאות מתייחסות להגדרות של מחבר ה-sink של BigQuery.

גודל התור

ההגדרה queueSize קובעת את המספר המקסימלי של בקשות כתיבה ל-BigQuery. כברירת מחדל, הערך הזה לא מוגדר, מה שעלול לגרום לשגיאות של מחסור בזיכרון באשכול Connect. מומלץ להגדיר את הערך הזה כדי להפחית את הסיכוי לשגיאות כאלה. הפריטים בתור הם בקשות כתיבה, והם לא תואמים ישירות למספר ההודעות או הבייטים. לכן קשה לדעת מראש מה הערך המומלץ.

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

סכימות למחבר BigQuery Sink

המחבר BigQuery Sink משתמש בממיר הערכים שהוגדר (value.converter) כדי לנתח ערכי רשומות של Kafka לשדות. לאחר מכן, הוא כותב את השדות בעמודות עם אותו שם בטבלה ב-BigQuery.

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

סכימה מבוססת-הודעה: הסכימה נכללת כחלק מכל הודעה.
סכימה מבוססת-טבלה: המחבר מסיק את סכימת ההודעה מסכימת הטבלה ב-BigQuery.
מרשם סכימות: המחבר קורא את הסכימה ממרשם סכימות, כמו מרשם הסכימות של השירות המנוהל ל-Apache Kafka (גרסת טרום-השקה (Preview)).

בקטעים הבאים מפורטות האפשרויות האלה.

סכימה מבוססת-הודעות

במצב הזה, כל רשומת Kafka כוללת סכימת JSON. המחבר משתמש בסכימה כדי לכתוב את נתוני הרשומה כשורה בטבלה ב-BigQuery.

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

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true

ערך לדוגמה של רשומת Kafka:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "field": "user",
        "type": "string",
        "optional": false
      },
      {
        "field": "age",
        "type": "int64",
        "optional": false
      }
    ]
  },
  "payload": {
    "user": "userId",
    "age": 30
  }
}

אם טבלת היעד כבר קיימת, הסכימה של הטבלה ב-BigQuery צריכה להיות תואמת לסכימה של ההודעה המוטמעת. אם הערך הוא autoCreateTables=true, המחבר יוצר באופן אוטומטי את טבלת היעד לפי הצורך. מידע נוסף זמין במאמר יצירת טבלאות.

אם רוצים שהמחבר יעדכן את סכימת הטבלה ב-BigQuery כשסכימות ההודעות משתנות, מגדירים את allowNewBigQueryFields,‏ allowSchemaUnionization או allowBigQueryRequiredFieldRelaxation כ-true.

סכימה מבוססת-טבלה

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

דרישות:

הטבלה ב-BigQuery כבר צריכה להיות קיימת.
נתוני הרשומה של Kafka צריכים להיות תואמים לסכימת הטבלה.
במצב הזה אין תמיכה בעדכוני סכימה דינמיים על סמך הודעות נכנסות.

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

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
bigQueryPartitionDecorator=false

אם הטבלה ב-BigQuery משתמשת בחלוקה למחיצות לפי זמן עם חלוקה יומית למחיצות, הערך של bigQueryPartitionDecorator יכול להיות true. אחרת, צריך להגדיר את המאפיין הזה לערך false.

ערך לדוגמה של רשומת Kafka:

{
  "user": "userId",
  "age": 30
}

רישום סכימה

במצב הזה, כל רשומת Kafka מכילה נתוני Apache Avro, והסכימה של ההודעה מאוחסנת במרשם סכימות.

כדי להשתמש במחבר BigQuery Sink עם מאגר סכימות, צריך להגדיר את המאפיינים הבאים במחבר:

value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=SCHEMA_REGISTRY_URL

מחליפים את SCHEMA_REGISTRY_URL בכתובת ה-URL של מאגר הסכימות.

כדי להשתמש במחבר עם רישום הסכימה של השירות המנוהל ל-Apache Kafka, מגדירים את המאפיין הבא:

value.converter.bearer.auth.credentials.source=GCP

מידע נוסף זמין במאמר שימוש ב-Kafka Connect עם רישום סכימה.

טבלאות מנוהלות של Apache Iceberg

המחבר BigQuery Sink תומך בטבלאות מנוהלות של Apache Iceberg (להלן, טבלאות מנוהלות של Iceberg) כיעד sink.

טבלאות מנוהלות של Iceberg מספקות את הבסיס לבניית מאגרי נתונים בפורמט פתוח ב- Google Cloud. טבלאות מנוהלות ב-Iceberg מציעות את אותה חוויה מנוהלת כמו טבלאות ב-BigQuery, אבל הנתונים מאוחסנים בקטגוריות אחסון בבעלות הלקוח באמצעות Parquet, כדי לאפשר יכולת פעולה הדדית עם פורמטים של טבלאות פתוחות ב-Apache Iceberg.

במאמר יצירת טבלת Apache Iceberg מוסבר איך יוצרים טבלת Apache Iceberg.

יצירת מחבר BigQuery Sink

המסוף

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

מעבר אל Connect Clusters
לוחצים על אשכול Connect שבו רוצים ליצור את המחבר.
לוחצים על Create connector.
מזינים מחרוזת בשם המחבר.

הנחיות למתן שמות למשאבים של שירות מנוהל ל-Apache Kafka
בשדה Connector plugin, בוחרים באפשרות BigQuery Sink.
בקטע Topics, מציינים את נושאי Kafka שמהם רוצים לקרוא. אפשר לציין רשימה של נושאים או ביטוי רגולרי שיתאים לשמות של נושאים.
- אפשרות 1: בוחרים באפשרות Select a list of Kafka topics. ברשימה Kafka topics, בוחרים נושא אחד או יותר. לוחצים על אישור.
- אפשרות 2: בוחרים באפשרות שימוש בביטוי רגולרי של נושא. בשדה Topic regex מזינים ביטוי רגולרי.
לוחצים על Dataset ומציינים מערך נתונים ב-BigQuery. אפשר לבחור מערך נתונים קיים או ליצור מערך נתונים חדש.
אופציונלי: בתיבה Configurations, מוסיפים מאפייני הגדרות או עורכים את מאפייני ברירת המחדל. מידע נוסף זמין במאמר הגדרת המחבר.
בוחרים את מדיניות ההפעלה מחדש של המשימה. למידע נוסף, קראו את המאמר מדיניות ההפעלה מחדש של משימות.
לוחצים על יצירה.

gcloud

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

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

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

‫CONNECTOR_ID: המזהה או השם של המחבר.
‫CONNECT_CLUSTER_ID: המזהה של אשכול Connect שבו נוצר המחבר.
‫LOCATION: המיקום של אשכול Connect.
‫CONFIG_FILE: הנתיב לקובץ תצורה מקומי בפורמט YAML או JSON.

דוגמה לקובץ תצורה של מחבר BigQuery Sink:

name: "BQ_SINK_CONNECTOR_ID"
project: "GCP_PROJECT_ID"
topics: "GMK_TOPIC_ID"
tasks.max: 3
connector.class: "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
defaultDataset: "BQ_DATASET_ID"

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

‫BQ_SINK_CONNECTOR_ID: המזהה או השם של מחבר ה-Sink של BigQuery. במאמר הנחיות למתן שמות למשאבים של שירות מנוהל ל-Apache Kafka מוסבר איך נותנים שמות למחברים. אי אפשר לשנות את השם של המחבר.
‫GCP_PROJECT_ID: המזהה של הפרויקט ב- Google Cloudשבו נמצא מערך הנתונים ב-BigQuery.
‫GMK_TOPIC_ID: המזהה של הנושא בשירות המנוהל ל-Apache Kafka, שממנו הנתונים זורמים למחבר BigQuery Sink.
‫BQ_DATASET_ID: המזהה של מערך הנתונים ב-BigQuery שמשמש ככיור של צינור עיבוד הנתונים.

Terraform

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

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

  configs = {
    "name"                           = "my-bigquery-sink-connector"
    "project"                        = data.google_project.default.project_id
    "topics"                         = "GMK_TOPIC_ID"
    "tasks.max"                      = "3"
    "connector.class"                = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector"
    "key.converter"                  = "org.apache.kafka.connect.storage.StringConverter"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "defaultDataset"                 = "BQ_DATASET_ID"
  }

  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"
)

// createBigQuerySinkConnector creates a BigQuery Sink connector.
func createBigQuerySinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, tasksMax, keyConverter, valueConverter, valueConverterSchemasEnable, defaultDataset 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 := "BQ_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// tasksMax := "3"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// defaultDataset := "BQ_DATASET_ID"
	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)

	// BigQuery Sink sample connector configuration
	config := map[string]string{
		"name":                           connectorID,
		"project":                        projectID,
		"topics":                         topics,
		"tasks.max":                      tasksMax,
		"connector.class":                "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
		"key.converter":                  keyConverter,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"defaultDataset":                 defaultDataset,
	}

	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 BigQuery sink 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 CreateBigQuerySinkConnector {

  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-bigquery-sink-connector";
    String bigqueryProjectId = "my-bigquery-project-id";
    String datasetName = "my-dataset";
    String kafkaTopicName = "kafka-topic";
    String maxTasks = "3";
    String connectorClass = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    createBigQuerySinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bigqueryProjectId,
        datasetName,
        kafkaTopicName,
        maxTasks,
        connectorClass,
        keyConverter,
        valueConverter,
        valueSchemasEnable);
  }

  public static void createBigQuerySinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bigqueryProjectId,
      String datasetName,
      String kafkaTopicName,
      String maxTasks,
      String connectorClass,
      String keyConverter,
      String valueConverter,
      String valueSchemasEnable)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("name", connectorId);
    configMap.put("project", bigqueryProjectId);
    configMap.put("topics", kafkaTopicName);
    configMap.put("tasks.max", maxTasks);
    configMap.put("connector.class", connectorClass);
    configMap.put("key.converter", keyConverter);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    configMap.put("defaultDataset", datasetName);

    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 BigQuery Sink 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.

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

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 = {
    "name": connector_id,
    "project": project_id,
    "topics": topics,
    "tasks.max": tasks_max,
    "connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
    "key.converter": key_converter,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "defaultDataset": default_dataset,
}

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}")

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

הגדרת המחבר

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

שם טבלה

כברירת מחדל, המחבר משתמש בשם הנושא כשם הטבלה ב-BigQuery. כדי להשתמש בשם אחר לטבלה, מגדירים את המאפיין topic2TableMap בפורמט הבא:

topic2TableMap=TOPIC_1:TABLE_1,TOPIC_2:TABLE_2,...

יצירת טבלאות

המחבר BigQuery Sink יכול ליצור את טבלאות היעד אם הן לא קיימות.

אם הערך הוא autoCreateTables=true, המחבר מנסה ליצור טבלאות ב-BigQuery שלא קיימות. ההגדרה הזו היא פעולת ברירת המחדל.
אם הערך הוא autoCreateTables=false, המחבר לא יוצר טבלאות. אם טבלת היעד לא קיימת, תופיע הודעת שגיאה.

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

allBQFieldsNullable
clusteringPartitionFieldNames
convertDoubleSpecialValues
partitionExpirationMs
sanitizeFieldNames
sanitizeTopics
timestampPartitionFieldName

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

מטא-נתונים של Kafka

כדי למפות נתונים נוספים מ-Kafka, כמו מטא-נתונים ומידע על מפתחות, לטבלה ב-BigQuery, צריך להגדיר את השדות kafkaDataFieldName ו-kafkaKeyFieldName בהתאמה. דוגמאות לפרטי מטא-נתונים: הנושא, המחיצה, ההיסט וזמן ההוספה של Kafka.

מה השלב הבא?

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

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

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

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

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

מתן הרשאות כתיבה בטבלה ב-BigQuery

פרמטרים מרכזיים להגדרה

גודל התור

סכימות למחבר BigQuery Sink

סכימה מבוססת-הודעות

סכימה מבוססת-טבלה

רישום סכימה

טבלאות מנוהלות של Apache Iceberg

יצירת מחבר BigQuery Sink

המסוף

gcloud

Terraform

המשך

Java

Python

הגדרת המחבר

שם טבלה

יצירת טבלאות

מטא-נתונים של Kafka

מה השלב הבא?

יצירת מחבר BigQuery Sink