Cloud Storage 싱크 커넥터 만들기

Cloud Storage 싱크 커넥터를 사용하면 Kafka 주제에서 Cloud Storage 버킷으로 데이터를 스트리밍할 수 있습니다. 이는 비용 효율적이고 확장 가능한 방식으로 대량의 데이터를 저장하고 처리하는 데 유용합니다.

시작하기 전에

Cloud Storage 싱크 커넥터를 만들기 전에 다음을 확인하세요.

• Connect 클러스터의 Apache Kafka용 관리형 서비스 클러스터를 만듭니다. Connect 클러스터와 연결된 기본 Kafka 클러스터입니다. 커넥터 파이프라인의 한쪽 끝을 형성하는 소스 클러스터이기도 합니다.

• Cloud Storage 싱크 커넥터를 호스팅할 Connect 클러스터를 만듭니다.

• Kafka에서 스트리밍된 데이터를 저장할 Cloud Storage 버킷을 만듭니다.

• 소스 클러스터 내에서 Kafka 주제를 만들고 구성합니다. 데이터가 이 Kafka 주제에서 대상 Cloud Storage 버킷으로 이동합니다.

필수 역할 및 권한

Cloud Storage 싱크 커넥터를 만드는 데 필요한 권한을 얻으려면 관리자에게 프로젝트에 대한 관리형 Kafka 커넥터 편집자 (roles/managedkafka.connectorEditor) IAM 역할을 부여해 달라고 요청하세요. 역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

이 사전 정의된 역할에는 Cloud Storage 싱크 커넥터를 만드는 데 필요한 권한이 포함되어 있습니다. 필요한 정확한 권한을 보려면 필수 권한 섹션을 펼치세요.

필수 권한

Cloud Storage 싱크 커넥터를 만들려면 다음 권한이 필요합니다.

• 상위 연결 클러스터에 커넥터 생성 권한을 부여합니다. managedkafka.connectors.create

커스텀 역할이나 다른 사전 정의된 역할을 사용하여 이 권한을 부여받을 수도 있습니다.

관리형 Kafka 커넥터 편집자 역할에 대한 자세한 내용은 Apache Kafka용 관리형 서비스 사전 정의된 역할을 참고하세요.

Apache Kafka용 관리형 서비스 클러스터가 Connect 클러스터와 동일한 프로젝트에 있는 경우 추가 권한이 필요하지 않습니다. Connect 클러스터가 다른 프로젝트에 있는 경우 다른 프로젝트에서 Connect 클러스터 만들기를 참고하세요.

Cloud Storage 버킷에 쓸 수 있는 권한 부여

service-<project_number>@gcp-sa-managedkafka.iam.gserviceaccount.com 형식을 따르는 Connect 클러스터 서비스 계정에는 다음 Cloud Storage 권한이 필요합니다.

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

이렇게 하려면 Cloud Storage 버킷이 포함된 프로젝트에서 Connect 클러스터 서비스 계정에 스토리지 객체 사용자 (roles/storage.objectUser) 역할을 부여하세요.

Cloud Storage 싱크 커넥터 작동 방식

Cloud Storage 싱크 커넥터는 하나 이상의 Kafka 주제에서 데이터를 가져와 단일 Cloud Storage 버킷 내의 객체에 데이터를 씁니다.

Cloud Storage 싱크 커넥터가 데이터를 복사하는 방식에 대한 자세한 내용은 다음과 같습니다.

• 커넥터는 소스 클러스터 내의 하나 이상의 Kafka 주제에서 메시지를 소비합니다.

• 커넥터는 커넥터 구성에 지정된 타겟 Cloud Storage 버킷에 데이터를 씁니다.

• 커넥터는 커넥터 구성의 특정 속성을 참조하여 Cloud Storage 버킷에 데이터를 쓸 때 데이터를 포맷합니다. 기본적으로 출력 파일은 CSV 형식입니다. format.output.type 속성을 구성하여 JSON과 같은 다양한 출력 형식을 지정할 수 있습니다.

• 커넥터는 Cloud Storage 버킷에 작성되는 파일의 이름도 지정합니다. file.name.prefix 및 file.name.template 속성을 사용하여 파일 이름을 맞춤설정할 수 있습니다. 예를 들어 파일 이름에 Kafka 주제 이름이나 메시지 키를 포함할 수 있습니다.

• Kafka 레코드에는 헤더, 키, 값의 세 가지 구성요소가 있습니다.

  • format.output.fields를 설정하여 헤더를 포함하면 출력 파일에 헤더를 포함할 수 있습니다. 예를 들면 format.output.fields=value,headers입니다.

  • key을 포함하도록 format.output.fields을 설정하여 출력 파일에 키를 포함할 수 있습니다. 예를 들면 format.output.fields=key,value,headers입니다.

    키는 file.name.template 속성에 key를 포함하여 레코드를 그룹화하는 데도 사용할 수 있습니다.

• format.output.fields이 기본적으로 value로 설정되어 있으므로 기본적으로 출력 파일에 값을 포함할 수 있습니다.

• 커넥터는 변환되고 형식이 지정된 데이터를 지정된 Cloud Storage 버킷에 씁니다.

• file.compression.type 속성을 사용하여 파일 압축을 구성하면 커넥터가 Cloud Storage 버킷에 저장된 파일을 압축합니다.

• 변환기 구성은 format.output.type 속성으로 제한됩니다.

  • 예를 들어 format.output.type이 csv로 설정된 경우 키 변환기는 org.apache.kafka.connect.converters.ByteArrayConverter 또는 org.apache.kafka.connect.storage.StringConverter이어야 하고 값 변환기는 org.apache.kafka.connect.converters.ByteArrayConverter이어야 합니다.

  • format.output.type이 json로 설정된 경우 value.converter.schemas.enable 속성이 true이더라도 값 및 키 스키마가 출력 파일의 데이터와 함께 작성되지 않습니다.

• tasks.max 속성은 커넥터의 동시 로드 수준을 제어합니다. tasks.max를 늘리면 처리량이 향상될 수 있지만 실제 병렬 처리는 Kafka 주제의 파티션 수에 의해 제한됩니다.

Cloud Storage 싱크 커넥터의 속성

Cloud Storage 싱크 커넥터를 만들 때 다음 속성을 지정합니다.

커넥터 이름

커넥터의 이름 또는 ID입니다. 리소스 이름을 지정하는 방법에 대한 가이드라인은 Apache Kafka용 관리형 서비스 리소스 이름 지정 가이드라인을 참고하세요. 이름은 변경할 수 없습니다.

커넥터 플러그인 유형

Google Cloud 콘솔에서 커넥터 플러그인 유형으로 Cloud Storage 싱크를 선택합니다. 사용자 인터페이스를 사용하여 커넥터를 구성하지 않는 경우 커넥터 클래스도 지정해야 합니다.

주제

커넥터가 메시지를 소비하는 Kafka 주제입니다. 하나 이상의 주제를 지정하거나 정규 표현식을 사용하여 여러 주제를 일치시킬 수 있습니다. 예를 들어 'topic'으로 시작하는 모든 주제를 일치시키려면 topic.*를 사용합니다. 이러한 주제는 Connect 클러스터와 연결된 Apache Kafka용 관리형 서비스 클러스터 내에 있어야 합니다.

Cloud Storage 버킷

데이터가 저장되는 Cloud Storage 버킷을 선택하거나 만듭니다.

구성

이 섹션에서는 Cloud Storage 싱크 커넥터의 추가 커넥터별 구성 속성을 지정할 수 있습니다.

Kafka 주제의 데이터는 Avro, JSON, 원시 바이트 등 다양한 형식일 수 있으므로 구성의 핵심 부분은 변환기를 지정하는 것입니다. 변환기는 Kafka 주제에 사용된 형식의 데이터를 Kafka Connect의 표준화된 내부 형식으로 변환합니다. 그런 다음 Cloud Storage 싱크 커넥터가 이 내부 데이터를 가져와서 쓰기 전에 Cloud Storage 버킷에 필요한 형식으로 변환합니다.

Kafka Connect의 변환기 역할, 지원되는 변환기 유형, 일반적인 구성 옵션에 관한 일반적인 내용은 변환기를 참고하세요.

다음은 Cloud Storage 싱크 커넥터와 관련된 몇 가지 구성입니다.

• gcs.credentials.default: 실행 환경에서 Google Cloud 사용자 인증 정보를 자동으로 검색할지 여부입니다. true로 설정해야 합니다.

• gcs.bucket.name: 데이터가 작성되는 Cloud Storage 버킷의 이름을 지정합니다. 설정해야 합니다.

• file.compression.type: Cloud Storage 버킷에 저장된 파일의 압축 유형을 설정합니다. 예를 들어 gzip, snappy, zstd, none가 있습니다. 기본값은 none입니다.

• file.name.prefix: Cloud Storage 버킷에 저장된 각 파일의 이름에 추가할 접두사입니다. 기본값은 비어 있습니다.

• format.output.type: Cloud Storage 출력 파일에 데이터를 쓰는 데 사용되는 데이터 형식의 유형입니다. 지원되는 값은 csv, json, jsonl, parquet입니다. 기본값은 csv입니다.

이 커넥터에 사용할 수 있는 구성 속성 목록은 Cloud Storage 싱크 커넥터 구성을 참고하세요.

Cloud Storage 싱크 커넥터 만들기

커넥터를 만들기 전에 Cloud Storage 싱크 커넥터의 속성 문서를 검토하세요.

콘솔

1. Google Cloud 콘솔에서 클러스터 연결 페이지로 이동합니다.

   클러스터 연결로 이동

2. 커넥터를 만들려는 Connect 클러스터를 클릭합니다.

   클러스터 연결 세부정보 페이지가 표시됩니다.

3. 커넥터 만들기를 클릭합니다.

   Kafka 커넥터 만들기 페이지가 표시됩니다.

4. 커넥터 이름에 문자열을 입력합니다.

   커넥터 이름을 지정하는 방법에 대한 가이드라인은 Apache Kafka용 관리형 서비스 리소스 이름 지정 가이드라인을 참고하세요.

5. 커넥터 플러그인에서 Cloud Storage 싱크를 선택합니다.

6. 데이터를 스트리밍할 수 있는 주제를 지정합니다.

7. 데이터를 저장할 스토리지 버킷을 선택합니다.

8. (선택사항) 구성 섹션에서 추가 설정을 구성합니다.

9. 작업 재시작 정책을 선택합니다. 자세한 내용은 작업 다시 시작 정책을 참고하세요.

10. 만들기를 클릭합니다.

gcloud

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. 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: 커넥터의 ID 또는 이름입니다. 커넥터 이름을 지정하는 방법에 대한 가이드라인은 Apache Kafka용 관리형 서비스 리소스 이름 지정 가이드라인을 참고하세요. 커넥터의 이름은 변경할 수 없습니다.

   • LOCATION: 커넥터를 만드는 위치입니다. 이 위치는 Connect 클러스터를 만든 위치와 동일해야 합니다.

   • CONNECT_CLUSTER_ID: 커넥터가 생성된 Connect 클러스터의 ID입니다.

   • CONFIG_FILE: BigQuery 싱크 커넥터의 YAML 구성 파일 경로입니다.

   다음은 Cloud Storage 싱크 커넥터의 구성 파일의 예입니다.

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

   다음을 바꿉니다.

   • GMK_TOPIC_ID: 데이터가 Cloud Storage 싱크 커넥터로 흐르는 Apache Kafka용 관리형 서비스 주제의 ID입니다.

   • GCS_BUCKET_NAME: 파이프라인의 싱크 역할을 하는 Cloud Storage 버킷의 이름입니다.

   • GCS_SINK_CONNECTOR_ID: Cloud Storage 싱크 커넥터의 ID 또는 이름입니다. 커넥터 이름을 지정하는 방법에 대한 가이드라인은 Apache Kafka용 관리형 서비스 리소스 이름 지정 가이드라인을 참고하세요. 커넥터의 이름은 변경할 수 없습니다.

Terraform

Terraform 리소스를 사용하여 커넥터를 만들 수 있습니다.

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
}

Terraform 구성을 적용하거나 삭제하는 방법은 기본 Terraform 명령어를 참조하세요.

Go

이 샘플을 사용해 보기 전에 클라이언트 라이브러리 설치의 Go 설정 안내를 따르세요. 자세한 내용은 Apache Kafka용 관리형 서비스 Go API 참조 문서를 참고하세요.

Managed Service for Apache Kafka에 인증하려면 애플리케이션 기본 사용자 인증 정보(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"
)

// 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 설정 안내를 따르세요. 자세한 내용은 Apache Kafka용 관리형 서비스 Java API 참조 문서를 참고하세요.

Managed Service for Apache Kafka에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 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 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

이 샘플을 시도하기 전에 클라이언트 라이브러리 설치의 Python 설정 안내를 따르세요. 자세한 내용은 Apache Kafka용 관리형 서비스 Python API 참조 문서를 참고하세요.

Managed Service for Apache Kafka에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 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 = {
    "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}")

커넥터를 만든 후에는 커넥터를 수정, 삭제, 일시중지, 중지 또는 다시 시작할 수 있습니다.

다음 단계

• Cloud Storage 싱크 커넥터 문제 해결

• 커넥터 일시중지, 재개, 중지, 다시 시작

• 모든 커넥터 나열

• 커넥터 세부정보 보기

• 커넥터 업데이트

Apache Kafka®는 미국 및/또는 다른 국가에서 사용되는 Apache Software Foundation 또는 해당 계열사의 등록 상표입니다.