Kafka コネクタを使用してデータをロードする

StarRocks は、Apache Kafka® コネクタ (StarRocks Connector for Apache Kafka®) という独自開発のコネクタを提供しており、Kafka からメッセージを継続的に消費し、それを StarRocks にロードします。Kafka コネクタは、少なくとも一度のセマンティクスを保証します。

Kafka コネクタは Kafka Connect とシームレスに統合でき、StarRocks が Kafka エコシステムとより良く統合されます。リアルタイムデータを StarRocks にロードしたい場合には賢明な選択です。Routine Load と比較して、以下のシナリオでは Kafka コネクタの使用が推奨されます。

• Routine Load は CSV、JSON、Avro フォーマットでのデータロードのみをサポートしていますが、Kafka コネクタは Protobuf など、より多くのフォーマットでのデータロードが可能です。Kafka Connect のコンバータを使用してデータを JSON や CSV フォーマットに変換できれば、Kafka コネクタを介して StarRocks にデータをロードできます。
• Debezium フォーマットの CDC データなど、データ変換をカスタマイズします。
• 複数の Kafka トピックからデータをロードします。
• Confluent Cloud からデータをロードします。
• ロードバッチサイズ、並行性、その他のパラメータを細かく制御して、ロード速度とリソース利用のバランスを取る必要があります。

準備

バージョン要件

コネクタ	Kafka	StarRocks	Java
1.0.4	3.4	2.5 and later	8
1.0.3	3.4	2.5 and later	8

Kafka 環境のセットアップ

自己管理の Apache Kafka クラスターと Confluent Cloud の両方がサポートされています。

• 自己管理の Apache Kafka クラスターの場合、Apache Kafka クイックスタート を参照して、Kafka クラスターを迅速にデプロイできます。Kafka Connect はすでに Kafka に統合されています。
• Confluent Cloud の場合、Confluent アカウントを持ち、クラスターを作成していることを確認してください。

Kafka コネクタのダウンロード

Kafka コネクタを Kafka Connect に提出します。

• 自己管理の Kafka クラスター:

  starrocks-kafka-connector-xxx.tar.gz をダウンロードして解凍します。

• Confluent Cloud:

  現在、Kafka コネクタは Confluent Hub にアップロードされていません。starrocks-kafka-connector-xxx.tar.gz をダウンロードして解凍し、ZIP ファイルにパッケージして Confluent Cloud にアップロードする必要があります。

ネットワーク構成

Kafka が配置されているマシンが StarRocks クラスターの FE ノードに http_port (デフォルト: 8030) および query_port (デフォルト: 9030) を介してアクセスでき、BE ノードに be_http_port (デフォルト: 8040) を介してアクセスできることを確認してください。

使用方法

このセクションでは、自己管理の Kafka クラスターを例にとり、Kafka コネクタと Kafka Connect を設定し、Kafka Connect を実行して StarRocks にデータをロードする方法を説明します。

データセットの準備

Kafka クラスターのトピック test に JSON フォーマットのデータが存在すると仮定します。

{"id":1,"city":"New York"}
{"id":2,"city":"Los Angeles"}
{"id":3,"city":"Chicago"}

テーブルの作成

StarRocks クラスターのデータベース example_db に JSON フォーマットデータのキーに基づいてテーブル test_tbl を作成します。

CREATE DATABASE example_db;
USE example_db;
CREATE TABLE test_tbl (id INT, city STRING);

Kafka コネクタと Kafka Connect の設定と実行、データのロード

スタンドアロンモードで Kafka Connect を実行

1. Kafka コネクタを設定します。Kafka インストールディレクトリの config ディレクトリに、Kafka コネクタ用の設定ファイル connect-StarRocks-sink.properties を作成し、以下のパラメータを設定します。詳細なパラメータと説明については、Parameters を参照してください。

   注記

   Kafka コネクタはシンクコネクタです。

   name=starrocks-kafka-connector
   connector.class=com.starrocks.connector.kafka.StarRocksSinkConnector
   topics=test
   key.converter=org.apache.kafka.connect.json.JsonConverter
   value.converter=org.apache.kafka.connect.json.JsonConverter
   key.converter.schemas.enable=true
   value.converter.schemas.enable=false
   # StarRocks クラスター内の FE の HTTP URL。デフォルトポートは 8030 です。
   starrocks.http.url=192.168.xxx.xxx:8030
   # Kafka トピック名が StarRocks テーブル名と異なる場合、それらの間のマッピング関係を設定する必要があります。
   starrocks.topic2table.map=test:test_tbl
   # StarRocks ユーザー名を入力します。
   starrocks.username=user1
   # StarRocks パスワードを入力します。
   starrocks.password=123456
   starrocks.database.name=example_db
   sink.properties.strip_outer_array=true

   NOTICE

   ソースデータが Debezium フォーマットの CDC データであり、StarRocks テーブルが主キーテーブルである場合、ソースデータの変更を主キーテーブルに同期するために transform を設定する必要があります。

2. Kafka Connect を設定して実行します。

   1. Kafka Connect を設定します。config ディレクトリ内の設定ファイル config/connect-standalone.properties に以下のパラメータを設定します。詳細なパラメータと説明については、Running Kafka Connect を参照してください。以下の例では starrocks-kafka-connector バージョン 1.0.3 を使用しています。新しいバージョンを使用する場合は、対応する変更を行う必要があります。

      # Kafka ブローカーのアドレス。複数の Kafka ブローカーのアドレスはカンマ (,) で区切る必要があります。
      # この例では、Kafka クラスターにアクセスするためのセキュリティプロトコルとして PLAINTEXT を使用しています。他のセキュリティプロトコルを使用して Kafka クラスターにアクセスする場合は、このファイルに関連情報を設定する必要があります。
      bootstrap.servers=<kafka_broker_ip>:9092
      offset.storage.file.filename=/tmp/connect.offsets
      offset.flush.interval.ms=10000
      key.converter=org.apache.kafka.connect.json.JsonConverter
      value.converter=org.apache.kafka.connect.json.JsonConverter
      key.converter.schemas.enable=true
      value.converter.schemas.enable=false
      # 解凍後の starrocks-kafka-connector の絶対パス。例:
      plugin.path=/home/kafka-connect/starrocks-kafka-connector-1.0.3

   2. Kafka Connect を実行します。

      CLASSPATH=/home/kafka-connect/starrocks-kafka-connector-1.0.3/* bin/connect-standalone.sh config/connect-standalone.properties config/connect-starrocks-sink.properties

分散モードで Kafka Connect を実行

1. Kafka Connect を設定して実行します。

   1. Kafka Connect を設定します。config ディレクトリ内の設定ファイル config/connect-distributed.properties に以下のパラメータを設定します。詳細なパラメータと説明については、Running Kafka Connect を参照してください。

      # Kafka ブローカーのアドレス。複数の Kafka ブローカーのアドレスはカンマ (,) で区切る必要があります。
      # この例では、Kafka クラスターにアクセスするためのセキュリティプロトコルとして PLAINTEXT を使用しています。他のセキュリティプロトコルを使用して Kafka クラスターにアクセスする場合は、このファイルに関連情報を設定する必要があります。
      bootstrap.servers=<kafka_broker_ip>:9092
      offset.storage.file.filename=/tmp/connect.offsets
      offset.flush.interval.ms=10000
      key.converter=org.apache.kafka.connect.json.JsonConverter
      value.converter=org.apache.kafka.connect.json.JsonConverter
      key.converter.schemas.enable=true
      value.converter.schemas.enable=false
      # 解凍後の starrocks-kafka-connector の絶対パス。例:
      plugin.path=/home/kafka-connect/starrocks-kafka-connector-1.0.3

   2. Kafka Connect を実行します。

      CLASSPATH=/home/kafka-connect/starrocks-kafka-connector-1.0.3/* bin/connect-distributed.sh config/connect-distributed.properties

2. Kafka コネクタを設定して作成します。分散モードでは、REST API を通じて Kafka コネクタを設定して作成する必要があります。パラメータと説明については、Parameters を参照してください。

   注記

   Kafka コネクタはシンクコネクタです。

   curl -i http://127.0.0.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{
     "name":"starrocks-kafka-connector",
     "config":{
       "connector.class":"com.starrocks.connector.kafka.StarRocksSinkConnector",
       "topics":"test",
       "key.converter":"org.apache.kafka.connect.json.JsonConverter",
       "value.converter":"org.apache.kafka.connect.json.JsonConverter",
       "key.converter.schemas.enable":"true",
       "value.converter.schemas.enable":"false",
       "starrocks.http.url":"192.168.xxx.xxx:8030",
       "starrocks.topic2table.map":"test:test_tbl",
       "starrocks.username":"user1",
       "starrocks.password":"123456",
       "starrocks.database.name":"example_db",
       "sink.properties.strip_outer_array":"true"
     }
   }'

   備考

   ソースデータが Debezium フォーマットの CDC データであり、StarRocks テーブルが主キーテーブルである場合、ソースデータの変更を主キーテーブルに同期するために transform を設定する必要があります。

StarRocks テーブルのクエリ

ターゲット StarRocks テーブル test_tbl をクエリします。

MySQL [example_db]> select * from test_tbl;

+------+-------------+
| id   | city        |
+------+-------------+
|    1 | New York    |
|    2 | Los Angeles |
|    3 | Chicago     |
+------+-------------+
3 rows in set (0.01 sec)

上記の結果が返された場合、データは正常にロードされています。

パラメータ

name

必須: YES
デフォルト値:
説明: この Kafka コネクタの名前。Kafka Connect クラスター内のすべての Kafka コネクタ間でグローバルに一意である必要があります。例: starrocks-kafka-connector。

connector.class

必須: YES
デフォルト値:
説明: この Kafka コネクタのシンクで使用されるクラス。値を com.starrocks.connector.kafka.StarRocksSinkConnector に設定します。

topics

必須: YES
デフォルト値:
説明: 購読する1つ以上のトピックで、各トピックは StarRocks テーブルに対応します。デフォルトでは、StarRocks はトピック名が StarRocks テーブル名と一致すると仮定します。したがって、StarRocks はトピック名を使用してターゲットの StarRocks テーブルを決定します。topics または topics.regex (下記) のいずれかを選択して入力してください。ただし、StarRocks テーブル名がトピック名と異なる場合は、オプションの starrocks.topic2table.map パラメータ (下記) を使用してトピック名からテーブル名へのマッピングを指定します。

topics.regex

必須:
デフォルト値: 購読する1つ以上のトピックに一致する正規表現。詳細については topics を参照してください。topics.regex または topics (上記) のいずれかを選択して入力してください。
説明:

starrocks.topic2table.map

必須: NO
デフォルト値:
説明: トピック名が StarRocks テーブル名と異なる場合の StarRocks テーブル名とトピック名のマッピング。フォーマットは <topic-1>:<table-1>,<topic-2>:<table-2>,... です。

starrocks.http.url

必須: YES
デフォルト値:
説明: StarRocks クラスター内の FE の HTTP URL。フォーマットは <fe_host1>:<fe_http_port1>,<fe_host2>:<fe_http_port2>,... です。複数のアドレスはカンマ (,) で区切ります。例: 192.168.xxx.xxx:8030,192.168.xxx.xxx:8030。

starrocks.database.name

必須: YES
デフォルト値:
説明: StarRocks データベースの名前。

starrocks.username

必須: YES
デフォルト値:
説明: StarRocks クラスターアカウントのユーザー名。ユーザーは StarRocks テーブルに対する INSERT 権限を持っている必要があります。

starrocks.password

必須: YES
デフォルト値:
説明: StarRocks クラスターアカウントのパスワード。

key.converter

必須: NO
デフォルト値: Kafka Connect クラスターで使用されるキーコンバータ
説明: このパラメータは、シンクコネクタ (Kafka-connector-starrocks) のキーコンバータを指定し、Kafka データのキーをデシリアライズするために使用されます。デフォルトのキーコンバータは、Kafka Connect クラスターで使用されるものです。

value.converter

必須: NO
デフォルト値: Kafka Connect クラスターで使用される値コンバータ
説明: このパラメータは、シンクコネクタ (Kafka-connector-starrocks) の値コンバータを指定し、Kafka データの値をデシリアライズするために使用されます。デフォルトの値コンバータは、Kafka Connect クラスターで使用されるものです。

key.converter.schema.registry.url

必須: NO
デフォルト値:
説明: キーコンバータのスキーマレジストリ URL。

value.converter.schema.registry.url

必須: NO
デフォルト値:
説明: 値コンバータのスキーマレジストリ URL。

tasks.max

必須: NO
デフォルト値: 1
説明: Kafka コネクタが作成できるタスクスレッドの上限で、通常は Kafka Connect クラスターのワーカーノードの CPU コア数と同じです。このパラメータを調整してロードパフォーマンスを制御できます。

bufferflush.maxbytes

必須: NO
デフォルト値: 94371840(90M)
説明: 一度に StarRocks に送信される前にメモリに蓄積できるデータの最大サイズ。最大値は 64 MB から 10 GB の範囲です。Stream Load SDK バッファはデータをバッファリングするために複数の Stream Load ジョブを作成する可能性があることに注意してください。したがって、ここで言及されているしきい値は、総データサイズを指します。

bufferflush.intervalms

必須: NO
デフォルト値: 1000
説明: データのバッチを送信する間隔で、ロードの遅延を制御します。範囲: [1000, 3600000]。

connect.timeoutms

必須: NO
デフォルト値: 1000
説明: HTTP URL への接続のタイムアウト。範囲: [100, 60000]。

sink.properties.*

必須:
デフォルト値:
説明: ロード動作を制御するための Stream Load パラメータ。例えば、パラメータ sink.properties.format は Stream Load に使用されるフォーマット (CSV や JSON など) を指定します。サポートされているパラメータとその説明のリストについては、STREAM LOAD を参照してください。

sink.properties.format

必須: NO
デフォルト値: json
説明: Stream Load に使用されるフォーマット。Kafka コネクタは、データの各バッチをこのフォーマットに変換してから StarRocks に送信します。有効な値は csv と json です。詳細については、CSV パラメータ および JSON パラメータ を参照してください。

sink.properties.partial_update

必須: NO
デフォルト値: FALSE
説明: 部分更新を使用するかどうか。有効な値は TRUE と FALSE です。デフォルト値は FALSE で、この機能を無効にします。

sink.properties.partial_update_mode

必須: NO
デフォルト値: row
説明: 部分更新のモードを指定します。有効な値は row と column です。

• デフォルトの row は行モードでの部分更新を意味し、多くの列と小さなバッチでのリアルタイム更新に適しています。
• column は列モードでの部分更新を意味し、少ない列と多くの行でのバッチ更新に適しています。このようなシナリオでは、列モードを有効にすると更新速度が速くなります。例えば、100 列のテーブルで、すべての行に対して 10 列 (全体の 10%) のみが更新される場合、列モードの更新速度は 10 倍速くなります。

使用上の注意

フラッシュポリシー

Kafka コネクタはデータをメモリにバッファし、Stream Load を介して StarRocks にバッチでフラッシュします。以下の条件のいずれかが満たされた場合にフラッシュがトリガーされます。

• バッファされた行のバイト数が bufferflush.maxbytes の制限に達したとき。
• 最後のフラッシュからの経過時間が bufferflush.intervalms の制限に達したとき。
• タスクのオフセットをコミットしようとするコネクタの間隔に達したとき。この間隔は Kafka Connect の設定 offset.flush.interval.ms によって制御され、デフォルト値は 60000 です。

データの遅延を低くするために、これらの設定を Kafka コネクタの設定で調整します。ただし、フラッシュの頻度が増えると CPU と I/O の使用量が増加します。

制限

• Kafka トピックからの単一メッセージを複数のデータ行にフラット化して StarRocks にロードすることはサポートされていません。
• Kafka コネクタのシンクは少なくとも一度のセマンティクスを保証します。

ベストプラクティス

Debezium フォーマットの CDC データをロードする

Debezium は、さまざまなデータベースシステムでのデータ変更を監視し、これらの変更を Kafka にストリーミングすることをサポートする人気のある Change Data Capture (CDC) ツールです。以下の例では、PostgreSQL の変更を StarRocks の 主キーテーブル に書き込むために Kafka コネクタを設定して使用する方法を示します。

ステップ 1: Kafka のインストールと開始

NOTE

独自の Kafka 環境を持っている場合、このステップはスキップできます。

1. 公式サイトから最新の Kafka リリースをダウンロードし、パッケージを解凍します。

   tar -xzf kafka_2.13-3.7.0.tgz
   cd kafka_2.13-3.7.0

2. Kafka 環境を開始します。

   Kafka クラスター UUID を生成します。

   KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

   ログディレクトリをフォーマットします。

   bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

   Kafka サーバーを開始します。

   bin/kafka-server-start.sh config/kraft/server.properties

ステップ 2: PostgreSQL の設定

1. PostgreSQL ユーザーに REPLICATION 権限が付与されていることを確認します。

2. PostgreSQL の設定を調整します。

   postgresql.conf で wal_level を logical に設定します。

   wal_level = logical

   PostgreSQL サーバーを再起動して変更を適用します。

   pg_ctl restart

3. データセットを準備します。

   テーブルを作成し、テストデータを挿入します。

   CREATE TABLE customers (
     id int primary key ,
     first_name varchar(65533) NULL,
     last_name varchar(65533) NULL ,
     email varchar(65533) NULL 
   );
   
   INSERT INTO customers VALUES (1,'a','a','a@a.com');

4. Kafka で CDC ログメッセージを確認します。

   {
       "schema": {
           "type": "struct",
           "fields": [
               {
                   "type": "struct",
                   "fields": [
                       {
                           "type": "int32",
                           "optional": false,
                           "field": "id"
                       },
                       {
                           "type": "string",
                           "optional": true,
                           "field": "first_name"
                       },
                       {
                           "type": "string",
                           "optional": true,
                           "field": "last_name"
                       },
                       {
                           "type": "string",
                           "optional": true,
                           "field": "email"
                       }
                   ],
                   "optional": true,
                   "name": "test.public.customers.Value",
                   "field": "before"
               },
               {
                   "type": "struct",
                   "fields": [
                       {
                           "type": "int32",
                           "optional": false,
                           "field": "id"
                       },
                       {
                           "type": "string",
                           "optional": true,
                           "field": "first_name"
                       },
                       {
                           "type": "string",
                           "optional": true,
                           "field": "last_name"
                       },
                       {
                           "type": "string",
                           "optional": true,
                           "field": "email"
                       }
                   ],
                   "optional": true,
                   "name": "test.public.customers.Value",
                   "field": "after"
               },
               {
                   "type": "struct",
                   "fields": [
                       {
                           "type": "string",
                           "optional": false,
                           "field": "version"
                       },
                       {
                           "type": "string",
                           "optional": false,
                           "field": "connector"
                       },
                       {
                           "type": "string",
                           "optional": false,
                           "field": "name"
                       },
                       {
                           "type": "int64",
                           "optional": false,
                           "field": "ts_ms"
                       },
                       {
                           "type": "string",
                           "optional": true,
                           "name": "io.debezium.data.Enum",
                           "version": 1,
                           "parameters": {
                               "allowed": "true,last,false,incremental"
                           },
                           "default": "false",
                           "field": "snapshot"
                       },
                       {
                           "type": "string",
                           "optional": false,
                           "field": "db"
                       },
                       {
                           "type": "string",
                           "optional": true,
                           "field": "sequence"
                       },
                       {
                           "type": "string",
                           "optional": false,
                           "field": "schema"
                       },
                       {
                           "type": "string",
                           "optional": false,
                           "field": "table"
                       },
                       {
                           "type": "int64",
                           "optional": true,
                           "field": "txId"
                       },
                       {
                           "type": "int64",
                           "optional": true,
                           "field": "lsn"
                       },
                       {
                           "type": "int64",
                           "optional": true,
                           "field": "xmin"
                       }
                   ],
                   "optional": false,
                   "name": "io.debezium.connector.postgresql.Source",
                   "field": "source"
               },
               {
                   "type": "string",
                   "optional": false,
                   "field": "op"
               },
               {
                   "type": "int64",
                   "optional": true,
                   "field": "ts_ms"
               },
               {
                   "type": "struct",
                   "fields": [
                       {
                           "type": "string",
                           "optional": false,
                           "field": "id"
                       },
                       {
                           "type": "int64",
                           "optional": false,
                           "field": "total_order"
                       },
                       {
                           "type": "int64",
                           "optional": false,
                           "field": "data_collection_order"
                       }
                   ],
                   "optional": true,
                   "name": "event.block",
                   "version": 1,
                   "field": "transaction"
               }
           ],
           "optional": false,
           "name": "test.public.customers.Envelope",
           "version": 1
       },
       "payload": {
           "before": null,
           "after": {
               "id": 1,
               "first_name": "a",
               "last_name": "a",
               "email": "a@a.com"
           },
           "source": {
               "version": "2.5.3.Final",
               "connector": "postgresql",
               "name": "test",
               "ts_ms": 1714283798721,
               "snapshot": "false",
               "db": "postgres",
               "sequence": "[\"22910216\",\"22910504\"]",
               "schema": "public",
               "table": "customers",
               "txId": 756,
               "lsn": 22910504,
               "xmin": null
           },
           "op": "c",
           "ts_ms": 1714283798790,
           "transaction": null
       }
   }

ステップ 3: StarRocks の設定

PostgreSQL のソーステーブルと同じスキーマを持つ主キーテーブルを StarRocks に作成します。

CREATE TABLE `customers` (
  `id` int(11) COMMENT "",
  `first_name` varchar(65533) NULL COMMENT "",
  `last_name` varchar(65533) NULL COMMENT "",
  `email` varchar(65533) NULL COMMENT ""
) ENGINE=OLAP 
PRIMARY KEY(`id`) 
DISTRIBUTED BY hash(id) buckets 1
PROPERTIES (
"bucket_size" = "4294967296",
"in_memory" = "false",
"enable_persistent_index" = "false",
"replicated_storage" = "true",
"fast_schema_evolution" = "true"
);

ステップ 4: コネクタのインストール

1. コネクタをダウンロードし、plugins ディレクトリにパッケージを解凍します。

   mkdir plugins
   tar -zxvf debezium-debezium-connector-postgresql-2.5.3.zip -C plugins
   tar -zxvf starrocks-kafka-connector-1.0.3.tar.gz -C plugins

   このディレクトリは、config/connect-standalone.properties の設定項目 plugin.path の値です。

   plugin.path=/path/to/kafka_2.13-3.7.0/plugins

2. pg-source.properties に PostgreSQL ソースコネクタを設定します。

   {
     "name": "inventory-connector",
     "config": {
       "connector.class": "io.debezium.connector.postgresql.PostgresConnector", 
       "plugin.name": "pgoutput",
       "database.hostname": "localhost", 
       "database.port": "5432", 
       "database.user": "postgres", 
       "database.password": "", 
       "database.dbname" : "postgres", 
       "topic.prefix": "test"
     }
   }

3. sr-sink.properties に StarRocks シンクコネクタを設定します。

   {
       "name": "starrocks-kafka-connector",
       "config": {
           "connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
           "tasks.max": "1",
           "topics": "test.public.customers",
           "starrocks.http.url": "172.26.195.69:28030",
           "starrocks.database.name": "test",
           "starrocks.username": "root",
           "starrocks.password": "StarRocks@123",
           "sink.properties.strip_outer_array": "true",
           "connect.timeoutms": "3000",
           "starrocks.topic2table.map": "test.public.customers:customers",
           "transforms": "addfield,unwrap",
           "transforms.addfield.type": "com.starrocks.connector.kafka.transforms.AddOpFieldForDebeziumRecord",
           "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
           "transforms.unwrap.drop.tombstones": "true",
           "transforms.unwrap.delete.handling.mode": "rewrite"
       }
   }

   NOTE

   • StarRocks テーブルが主キーテーブルでない場合、addfield トランスフォームを指定する必要はありません。
   • unwrap トランスフォームは Debezium によって提供され、操作タイプに基づいて Debezium の複雑なデータ構造をアンラップするために使用されます。詳細については、New Record State Extraction を参照してください。

4. Kafka Connect を設定します。

   Kafka Connect の設定ファイル config/connect-standalone.properties に以下の設定項目を設定します。

   # Kafka ブローカーのアドレス。複数の Kafka ブローカーのアドレスはカンマ (,) で区切る必要があります。
   # この例では、Kafka クラスターにアクセスするためのセキュリティプロトコルとして PLAINTEXT を使用しています。
   # 他のセキュリティプロトコルを使用して Kafka クラスターにアクセスする場合は、この部分に関連情報を設定します。
   
   bootstrap.servers=<kafka_broker_ip>:9092
   offset.storage.file.filename=/tmp/connect.offsets
   key.converter=org.apache.kafka.connect.json.JsonConverter
   value.converter=org.apache.kafka.connect.json.JsonConverter
   key.converter.schemas.enable=true
   value.converter.schemas.enable=false
   
   # 解凍後の starrocks-kafka-connector の絶対パス。例:
   plugin.path=/home/kafka-connect/starrocks-kafka-connector-1.0.3
   
   # フラッシュポリシーを制御するパラメータ。詳細については、使用上の注意セクションを参照してください。
   offset.flush.interval.ms=10000
   bufferflush.maxbytes = xxx
   bufferflush.intervalms = xxx

   詳細なパラメータの説明については、Running Kafka Connect を参照してください。

ステップ 5: スタンドアロンモードで Kafka Connect を開始

スタンドアロンモードで Kafka Connect を実行してコネクタを起動します。

bin/connect-standalone.sh config/connect-standalone.properties config/pg-source.properties config/sr-sink.properties 

ステップ 6: データ取り込みの確認

以下の操作をテストし、データが正しく StarRocks に取り込まれていることを確認します。

INSERT

• PostgreSQL で:

postgres=# insert into customers values (2,'b','b','b@b.com');
INSERT 0 1
postgres=# select * from customers;
 id | first_name | last_name |  email  
----+------------+-----------+---------
  1 | a          | a         | a@a.com
  2 | b          | b         | b@b.com
(2 rows)

• StarRocks で:

MySQL [test]> select * from customers;
+------+------------+-----------+---------+
| id   | first_name | last_name | email   |
+------+------------+-----------+---------+
|    1 | a          | a         | a@a.com |
|    2 | b          | b         | b@b.com |
+------+------------+-----------+---------+
2 rows in set (0.01 sec)

UPDATE

• PostgreSQL で:

postgres=# update customers set email='c@c.com';
UPDATE 2
postgres=# select * from customers;
 id | first_name | last_name |  email  
----+------------+-----------+---------
  1 | a          | a         | c@c.com
  2 | b          | b         | c@c.com
(2 rows)

• StarRocks で:

MySQL [test]> select * from customers;
+------+------------+-----------+---------+
| id   | first_name | last_name | email   |
+------+------------+-----------+---------+
|    1 | a          | a         | c@c.com |
|    2 | b          | b         | c@c.com |
+------+------------+-----------+---------+
2 rows in set (0.00 sec)

DELETE

• PostgreSQL で:

postgres=# delete from customers where id=1;
DELETE 1
postgres=# select * from customers;
 id | first_name | last_name |  email  
----+------------+-----------+---------
  2 | b          | b         | c@c.com
(1 row)

• StarRocks で:

MySQL [test]> select * from customers;
+------+------------+-----------+---------+
| id   | first_name | last_name | email   |
+------+------------+-----------+---------+
|    2 | b          | b         | c@c.com |
+------+------------+-----------+---------+
1 row in set (0.00 sec)