Membuat konektor Cloud Storage Sink

Konektor Sink Cloud Storage mengalirkan data dari Kafka ke bucket Cloud Storage, sehingga Anda dapat menyimpan dan memproses data dalam jumlah besar secara hemat biaya dan skalabel.

Kasus penggunaan untuk konektor Sink Cloud Storage meliputi:

  • Penyerapan data lake. Menyimpan data Kafka di data lake untuk pengarsipan jangka panjang dan batch processing.

  • Mengarsipkan data untuk memenuhi persyaratan peraturan.

Sebelum memulai

Sebelum membuat konektor Sink Cloud Storage, pastikan Anda memiliki hal berikut:

Peran dan izin yang diperlukan

Untuk mendapatkan izin yang diperlukan guna membuat konektor, minta administrator untuk memberi Anda peran IAM Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) di project Anda. Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses ke project, folder, dan organisasi.

Peran bawaan ini berisi izin yang diperlukan untuk membuat konektor. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk membuat konektor:

  • Buat konektor: managedkafka.connectors.create

Anda mungkin juga bisa mendapatkan izin ini dengan peran khusus atau peran bawaan lainnya.

Memberikan izin untuk menulis ke bucket Cloud Storage

Akun layanan Managed Kafka harus memiliki izin berikut untuk menulis pesan ke bucket Cloud Storage:

  • storage.objects.create
  • storage.objects.delete

Berikan peran Storage Object User (roles/storage.objectUser) ke akun layanan di project yang berisi bucket Cloud Storage.

Akun layanan Managed Kafka memiliki format berikut: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, dengan PROJECT_NUMBER adalah nomor project cluster Connect.

Jika cluster Connect Anda berada di project yang berbeda dengan cluster Managed Service untuk Apache Kafka, lihat Membuat cluster Connect di project yang berbeda.

Membuat konektor Cloud Storage Sink

Untuk membuat konektor Cloud Storage Sink, lakukan langkah-langkah berikut.

Konsol

  1. Di konsol Google Cloud , buka halaman Connect Clusters.

    Buka Connect Clusters

  2. Klik cluster Connect yang konektornya ingin Anda buat.

    Halaman Connect cluster details akan ditampilkan.

  3. Klik Create connector.

    Halaman Create Kafka connector akan ditampilkan.

  4. Untuk nama konektor, masukkan string.

    Untuk panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka.

  5. Untuk Connector plugin, pilih Cloud Storage Sink.

  6. Tentukan Topik dari mana Anda dapat melakukan streaming data.

  7. Pilih Storage Bucket untuk menyimpan data.

  8. (Opsional) Konfigurasi setelan tambahan di bagian Configuration.

  9. Pilih Task restart policy. Untuk mengetahui informasi selengkapnya, lihat Kebijakan memulai ulang tugas.

  10. Klik Create.

gcloud

  1. Di konsol Google Cloud , aktifkan Cloud Shell.

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi pada sesi.

  2. Jalankan perintah gcloud managed-kafka connectors create:

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

    Ganti kode berikut:

    • CONNECTOR_ID: ID atau nama konektor. Untuk mengetahui panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka. Nama konektor tidak dapat diubah.

    • LOCATION: Lokasi tempat Anda membuat konektor. Lokasi ini harus sama dengan lokasi tempat Anda membuat cluster Connect.

    • CONNECT_CLUSTER_ID: ID cluster Connect tempat konektor dibuat.

    • CONFIG_FILE: Jalur ke file konfigurasi YAML untuk konektor.

    Berikut adalah contoh file konfigurasi untuk konektor Cloud Storage Sink:

    connector.class: "io.aiven.kafka.connect.gcs.GcsSinkConnector"
gcs.bucket.name: "BUCKET_NAME"
gcs.credentials.default: "true"
format.output.type: "json"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
topics: "TOPIC_LIST"
tasks.max: "3"

    Ganti kode berikut:

    • BUCKET_NAME: Nama bucket Cloud Storage yang akan ditulisi.

    • TOPIC_LIST: Daftar topik Kafka yang dipisahkan koma untuk dibaca. Daftar ini juga dapat menyertakan ekspresi reguler.

Terraform

Anda dapat menggunakan resource Terraform untuk membuat konektor.

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

  configs = {
    "connector.class"                = "io.aiven.kafka.connect.gcs.GcsSinkConnector"
    "tasks.max"                      = "3"
    "topics"                         = "GMK_TOPIC_ID"
    "gcs.bucket.name"                = "GCS_BUCKET_NAME"
    "gcs.credentials.default"        = "true"
    "format.output.type"             = "json"
    "name"                           = "my-gcs-sink-connector"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "key.converter"                  = "org.apache.kafka.connect.storage.StringConverter"
  }
  provider = google-beta
}

Untuk mempelajari cara menerapkan atau menghapus konfigurasi Terraform, lihat Perintah dasar Terraform.

Go

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Go API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi(ADC). Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

import (
	"context"
	"fmt"
	"io"

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

// createCloudStorageSinkConnector creates a Cloud Storage Sink connector.
func createCloudStorageSinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, gcsBucketName, tasksMax, formatOutputType, valueConverter, valueConverterSchemasEnable, keyConverter, gcsCredentialsDefault 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 := "GCS_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// gcsBucketName := "GCS_BUCKET_NAME"
	// tasksMax := "3"
	// formatOutputType := "json"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// gcsCredentialsDefault := "true"
	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)

	config := map[string]string{
		"connector.class":                "io.aiven.kafka.connect.gcs.GcsSinkConnector",
		"tasks.max":                      tasksMax,
		"topics":                         topics,
		"gcs.bucket.name":                gcsBucketName,
		"gcs.credentials.default":        gcsCredentialsDefault,
		"format.output.type":             formatOutputType,
		"name":                           connectorID,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"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 Cloud Storage sink connector: %s\n", resp.Name)
	return nil
}

Java

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi API Java Managed Service untuk Apache Kafka.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.


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 CreateCloudStorageSinkConnector {

  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-gcs-sink-connector";
    String bucketName = "my-gcs-bucket";
    String kafkaTopicName = "kafka-topic";
    String connectorClass = "io.aiven.kafka.connect.gcs.GcsSinkConnector";
    String maxTasks = "3";
    String gcsCredentialsDefault = "true";
    String formatOutputType = "json";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createCloudStorageSinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bucketName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        gcsCredentialsDefault,
        formatOutputType,
        valueConverter,
        valueSchemasEnable,
        keyConverter);
  }

  public static void createCloudStorageSinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bucketName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String gcsCredentialsDefault,
      String formatOutputType,
      String valueConverter,
      String valueSchemasEnable,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("tasks.max", maxTasks);
    configMap.put("topics", kafkaTopicName);
    configMap.put("gcs.bucket.name", bucketName);
    configMap.put("gcs.credentials.default", gcsCredentialsDefault);
    configMap.put("format.output.type", formatOutputType);
    configMap.put("name", connectorId);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    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 Cloud Storage Sink connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Python API Managed Service untuk Apache Kafka.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

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": "io.aiven.kafka.connect.gcs.GcsSinkConnector",
    "tasks.max": tasks_max,
    "topics": topics,
    "gcs.bucket.name": gcs_bucket_name,
    "gcs.credentials.default": "true",
    "format.output.type": format_output_type,
    "name": connector_id,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "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}")

Setelah membuat konektor, Anda dapat mengedit, menghapus, menjeda, menghentikan, atau memulai ulang konektor.

Mengonfigurasi konektor

Untuk mengetahui daftar lengkap properti yang khusus untuk konektor ini, lihat Konfigurasi dalam dokumentasi konektor Sink Cloud Storage.

Kolom output

Secara default, konektor hanya menulis nilai data dari setiap data Kafka. Untuk menyertakan kunci atau header data, tetapkan properti format.output.fields ke daftar kolom yang dipisahkan koma untuk disertakan. Kolom yang didukung adalah key, value, headers, dan timestamp. Contoh:

format.output.fields=key,value,headers

Format output

Konektor ini mendukung penulisan data Kafka dalam format CSV, JSON, dan Parquet. Bagian berikut menjelaskan cara mengonfigurasi format output.

Format CSV

Untuk menulis file CSV ke Cloud Storage, tetapkan format.output.type=csv. Gunakan pengonversi berikut:

  • Pengonversi kunci: org.apache.kafka.connect.storage.StringConverter atau org.apache.kafka.connect.json.JsonConverter

  • Konverter nilai: org.apache.kafka.connect.converters.ByteArrayConverter

Format JSON

Untuk menulis file berformat JSON ke Cloud Storage, tetapkan format.output.type ke salah satu nilai berikut:

  • json: Setiap file output berisi satu array objek JSON, dengan setiap objek JSON mewakili satu data Kafka.

  • jsonl: Konektor menulis setiap kumpulan data Kafka sebagai objek JSON pada baris terpisah.

Contoh berikut menunjukkan cara konektor menulis rekaman Kafka dalam format JSON:

{"headers":[{"key":"k1","value":"v1"},{"key":"k2","value":"v2"}],"value":"Hello world","key":"my-key"}

Jika Anda menyetel konverter kunci atau nilai ke org.apache.kafka.connect.converters.ByteArrayConverter, maka konektor menulis kolom tersebut sebagai nilai yang dienkode base64.

Konektor tidak menyertakan skema JSON dalam output, meskipun value.converter.schemas.enable adalah true.

Format Parquet

Untuk menulis file Parquet ke Cloud Storage, tetapkan format.output.type=parquet.

Jika konverter kunci atau nilai adalah org.apache.kafka.connect.json.JsonConverter, data JSON harus menyertakan skema. Tetapkan properti schemas.enable konverter ke true.

Untuk data JSON, konektor menulis data sebagai jenis struct dalam file Parquet. Misalnya, jika nilai rekaman Kafka memiliki kolom bernama userid, konektor akan menulis kolom bernama value.userid. Untuk meratakan nama kolom, tetapkan format.output.envelope=false. Jika format.output.envelope adalah false, format.output.fields harus menentukan satu kolom output.

Setelan file Cloud Storage

Setelan berikut mengontrol cara konektor menulis file ke Cloud Storage.

  • gcs.bucket.name: Wajib diisi. Menentukan nama bucket Cloud Storage tempat konektor menulis data.

  • file.compression.type: Menentukan jenis kompresi untuk file output. Contohnya adalah gzip, snappy, zstd, dan none. Nilai defaultnya adalah none.

  • file.name.prefix: Menentukan awalan yang akan ditambahkan ke nama setiap file. Nilai defaultnya kosong ("").

  • file.name.template: Menentukan template untuk nama file. Konektor ini mendukung variabel placeholder untuk template, seperti {{topic}} untuk nama topik dan {{partition}} untuk nomor partisi. Untuk mengetahui informasi selengkapnya, lihat dokumentasi konektor Cloud Storage Sink.

  • file.max.records: Jumlah maksimum data yang akan ditulis ke satu file. Jika nilainya 0, tidak ada nilai maksimum tertentu. Nilai defaultnya adalah 0.

  • file.max.bytes: Ukuran file maksimum yang akan ditulis, dalam byte. Jika nilainya 0, tidak ada maksimum tertentu. Nilai defaultnya adalah 0.

Pengelompokan kumpulan data

Saat konektor Cloud Storage Sink menulis rekaman Kafka, secara default konektor ini mengelompokkan rekaman menurut topik dan partisi. Misalnya, data dari partisi 0 topik my-topic ditulis ke file bernama my-topic-0-OFFSET, dengan OFFSET adalah offset awal.

Jika parameter file.name.template menyertakan variabel placeholder {{key}}, maka konektor akan mengelompokkan data menurut kunci. Dalam mode ini, konektor menulis file terpisah untuk setiap kunci catatan. Konektor akan mengganti setiap file dengan rekaman terbaru untuk kunci tersebut.

Kredensial

Properti gcs.credentials.default harus ditetapkan ke true. Setelan ini memungkinkan konektor menemukan kredensial Google Cloud secara otomatis.

Apa langkah selanjutnya?

Apache Kafka® adalah merek dagang terdaftar dari Apache Software Foundation atau afiliasinya di Amerika Serikat dan/atau negara lainnya.