1. Home
  2. Cloud
  3. Confluent Cloud への外部システムの接続

重要

このページの日本語コンテンツは古くなっている可能性があります。最新の英語版コンテンツをご覧になるには、こちらをクリックしてください。

Datagen Source Connector for Confluent Cloud

注釈

Confluent Platform 用にコネクターをローカルにインストールする場合は、「Datagen Source Connector for Confluent Platform」を参照してください。

Confluent Cloud Datagen Source Connectorを使用すると、開発およびテスト用に模擬データを生成できます。このコネクターは、Avro、JSON スキーマ、Protobuf、および JSON（スキーマレス）の出力フォーマットをサポートします。模擬ソースデータは、GitHub を介して datagen のリソース から提供されます。このコネクターは本稼働環境での使用には適していません。

Connect 用の Confluent Cloud API の使用に関する詳細とサンプルについては、「Confluent Cloud API for Connect」セクションを参照してください。

制限

以下の情報を確認してください。

クイックスタート

このクイックスタートを使用して、Confluent Cloud Datagen Source Connector の利用を開始することができます。このクイックスタートでは、コネクターを選択し、テストおよび開発で使用するためにコネクターを構成する基本的な方法について説明します。このコネクターは本稼働環境での使用には適していません。

前提条件
  • アマゾンウェブサービス （AWS）、Microsoft Azure （Azure）、または Google Cloud Platform （GCP）上の Confluent Cloud クラスターへのアクセスを許可されていること。
  • Confluent CLI がインストールされ、クラスター用に構成されていること。「Confluent CLI のインストール」を参照してください。
  • スキーマレジストリ ベースのフォーマット（Avro、JSON スキーマ、Protobuf など）を使用するには、Schema Registry を有効にしておく必要があります。詳細については、「スキーマレジストリ Enabled Environments」を参照してください。
  • Kafka クラスターの認証情報。次のいずれかの方法で認証情報を指定できます。
    • 既存の サービスアカウント のリソース ID を入力する。
    • コネクター用の Confluent Cloud サービスアカウント を作成します。「サービスアカウント」のドキュメントで、必要な ACL エントリを確認してください。一部のコネクターには固有の ACL 要件があります。
    • Confluent Cloud の API キーとシークレットを作成する。キーとシークレットを作成するには、confluent api-key create を使用するか、コネクターのセットアップ時に Cloud Console で直接 API キーとシークレットを自動生成します。

Confluent Cloud Console の使用

ステップ 1: Confluent Cloud クラスターを起動します。

インストール手順については、「Quick Start for Confluent Cloud」を参照してください。

ステップ 2: コネクターを追加します。

左のナビゲーションメニューの Data integration をクリックし、Connectors をクリックします。クラスター内に既にコネクターがある場合は、+ Add connector をクリックします。

ステップ 3: コネクターを選択します。

Click the Datagen Source connector card.

Datagen Source Connector Card

ステップ 4: コネクターの詳細情報を入力します。

注釈

  • すべての 前提条件 を満たしていることを確認してください。
  • アスタリスク（ * ）は必須項目であることを示しています。

Add Datagen Source Connector 画面で、以下を実行します。

Topics リストから、データの送信先のトピックを選択します。新しいトピックを作成するには、+Add new topic をクリックします。

ステップ 5: Kafka トピックを確認します。

コネクターが実行中になったら、メッセージが Kafka トピックに取り込まれていることを確認します。

Connect 用の Confluent Cloud API の使用に関する詳細とサンプルについては、「Confluent Cloud API for Connect」セクションを参照してください。

Confluent CLI の使用

以下の手順に従うと、Confluent CLI を使用してコネクターをセットアップし、実行できます。

注釈

ステップ 1: 使用可能なコネクターをリスト表示します。

以下のコマンドを入力して、使用可能なコネクターをリスト表示します。

confluent connect plugin list

ステップ 2: コネクターの必須の構成プロパティを表示します。

以下のコマンドを実行して、コネクターの必須プロパティを表示します。

confluent connect plugin describe <connector-catalog-name>

例:

confluent connect plugin describe DatagenSource

出力例:

Following are the required configs:
connector.class: DatagenSource
name
kafka.auth.mode
kafka.api.key
kafka.api.secret
kafka.topic
output.data.format
quickstart
tasks.max

ステップ 3: コネクターの構成ファイルを作成します。

コネクター構成プロパティを含む JSON ファイルを作成します。以下の例は、コネクターの必須プロパティを示しています。

{
    "name" : "<datagen-connector-name>",
    "connector.class": "DatagenSource",
    "kafka.auth.mode": "KAFKA_API_KEY",
    "kafka.api.key": "<my-kafka-api-key>",
    "kafka.api.secret" : "<my-kafka-api-secret>",
    "kafka.topic" : "topic1, topic2",
    "output.data.format" : "JSON",
    "quickstart" : "PAGEVIEWS",
    "tasks.max" : "1"
}

以下のプロパティ定義に注意してください。

  • "name": 新しいコネクターの名前を設定します。
  • "connector.class": コネクターのプラグイン名を指定します。

  • "kafka.auth.mode": 使用するコネクターの認証モードを指定します。オプションは SERVICE_ACCOUNT または KAFKA_API_KEY （デフォルト）です。API キーとシークレットを使用するには、構成プロパティ kafka.api.key と kafka.api.secret を構成例（前述）のように指定します。サービスアカウント を使用するには、プロパティ kafka.service.account.id=<service-account-resource-ID> に リソース ID を指定します。使用できるサービスアカウントのリソース ID のリストを表示するには、次のコマンドを使用します。

    confluent iam service-account list

    例:

    confluent iam service-account list

   Id     | Resource ID |       Name        |    Description
+---------+-------------+-------------------+-------------------
   123456 | sa-l1r23m   | sa-1              | Service account 1
   789101 | sa-l4d56p   | sa-2              | Service account 2

  • "kafka.topics": トピックを 1 つ入力するか、複数のトピックをコンマ区切りにして入力します。

  • "output.data.format": Kafka 出力レコード値のフォーマット（コネクターから送られるデータ）を設定します。指定可能なエントリは、AVRO、JSON_SR、PROTOBUF、または JSON （スキーマレス）です。スキーマベースのフォーマット（Avro など）を使用するには、Confluent Cloud Schema Registry を構成しておく必要があります。

  • "quickstart": 以下のいずれかのクイックスタートスキーマを入力します。

    • CLICKSTREAM
    • CLICSTREAM_CODES
    • CLICKSTREAM_USERS
    • INVENTORY
    • ORDERS
    • PAGEVIEWS（この例に示されているもの）
    • PRODUCT
    • RATINGS
    • STOCK_TRADES
    • USERS
    • USERS_ARRAY

    サンプルデータおよびスキーマの仕様を確認するには、datagen のリソース を参照してください。

Single Message Transforms: CLI を使用する SMT の追加の詳細については、Single Message Transforms（SMT） のドキュメントを参照してください。

すべてのプロパティの値と定義については、「構成プロパティ」を参照してください。

ステップ 4: 構成ファイルを読み込み、コネクターを作成します。

以下のコマンドを入力して、構成を読み込み、コネクターを起動します。

confluent connect create --config <file-name>.json

例:

confluent connect create --config datagen-source-config.json

出力例:

Created connector confluent-datagen-source lcc-ix4dl

ステップ 5: コネクターのステータスを確認します。

以下のコマンドを入力して、コネクターのステータスを確認します。

confluent connect list

出力例:

ID          |       Name               | Status  | Type
+-----------+--------------------------+---------+-------+
lcc-ix4dl   | confluent-datagen-source | RUNNING | source

ステップ 6: Kafka トピックを確認します。

コネクターが実行中になったら、メッセージが Kafka トピックに取り込まれていることを確認します。

Connect 用の Confluent Cloud API の使用に関する詳細とサンプルについては、「Confluent Cloud API for Connect」セクションを参照してください。

例

「Quick Start for Confluent Cloud」の手順に従い、Confluent Cloud 用の Datagen Source Connector を使用して Kafka にサンプルデータをストリーミングします。

構成プロパティ

このコネクターでは、以下のコネクター構成プロパティを使用します。

データへの接続方法（How should we connect to your data?）

name

コネクターの名前を設定します。

  • 型: string
  • 指定可能な値: 最大 64 文字の文字列
  • 重要度: 高

Kafka クラスターの認証情報（Kafka Cluster credentials）

kafka.auth.mode

Kafka の認証モード。KAFKA_API_KEY または SERVICE_ACCOUNT を指定できます。デフォルトは KAFKA_API_KEY モードです。

  • 型: string
  • デフォルト: KAFKA_API_KEY
  • 指定可能な値: KAFKA_API_KEY、SERVICE_ACCOUNT
  • 重要度: 高
kafka.api.key
  • 型: password
  • 重要度: 高
kafka.service.account.id

Kafka クラスターとの通信用の API キーを生成するために使用されるサービスアカウント。

  • 型: string
  • 重要度: 高
kafka.api.secret
  • 型: password
  • 重要度: 高

データの送信先トピック（Which topic do you want to send data to?）

kafka.topic

データの書き込み先トピック名を指定します。

  • 型: string
  • 重要度: 高

出力メッセージ（Output messages）

output.data.format

Kafka 出力レコード値のフォーマットを設定します。指定可能なエントリは、AVRO、JSON_SR、PROTOBUF、または JSON です。スキーマベースのメッセージフォーマット（AVRO、JSON_SR、PROTOBUF など）を使用する場合は、Confluent Cloud Schema Registry を構成しておく必要がある点に注意してください

  • 型: string
  • 重要度: 高
json.output.decimal.format

次の 2 つのリテラル値を取る Connect DECIMAL 論理型の値の JSON/JSON_SR シリアル化フォーマットを指定します。

BASE64 では、DECIMAL 論理型を base64 でエンコードされたバイナリデータとしてシリアル化します。

NUMERIC では、JSON/JSON_SR の Connect DECIMAL 論理型の値を、10 進数の値を表す数字としてシリアル化します。

  • 型: string
  • デフォルト: BASE64
  • 重要度: 低

Datagen の詳細（Datagen Details）

quickstart

Select from built-in quickstart schema specifications. Cannot be used with schema.string. Refer to kafka-connect-datagen on Github for additional information.

  • 型: string
  • Default: ""
  • 重要度: 高
schema.string

The literal JSON-encoded Avro schema to use. Cannot be set with quickstart.

  • 型: string
  • Default: ""
  • Valid Values: A string at most 10000 characters long
  • Importance: medium
schema.keyfield

Name of the field to use as message key. It's optional when using quickstart.

  • 型: string
  • Default: ""
  • Importance: medium
max.interval

メッセージ間の最大間隔（ミリ秒）を設定します。

  • 型: int
  • デフォルト: 1000
  • 重要度: 高

このコネクターのタスク数（Number of tasks for this connector）

tasks.max
  • 型: int
  • 指定可能な値: [1、…]
  • 重要度: 高

次のステップ

参考

フルマネージド型の Confluent Cloud コネクターが Confluent Cloud ksqlDB でどのように動作するかを示す例については、「Cloud ETL のデモ」を参照してください。この例では、Confluent CLI を使用して Confluent Cloud のリソースを管理する方法についても説明しています。

../_images/topology.ja.png