Tablestore:Kafka データのデータテーブルへの同期

• ワーカー設定ファイル connect-standalone.properties の例

  ワーカー設定には、Kafka の接続設定、シリアル化形式、オフセットのコミット頻度などのパラメーターが含まれます。 この例は、Kafka の公式ドキュメントに基づいています。 詳細については、「Kafka Connect」をご参照ください。

  # Licensed to the Apache Software Foundation (ASF) under one or more
  # contributor license agreements.  See the NOTICE file distributed with
  # this work for additional information regarding copyright ownership.
  # The ASF licenses this file to You under the Apache License, Version 2.0
  # (the "License"); you may not use this file except in compliance with
  # the License.  You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  # これらはデフォルト設定です。このファイルは、いくつかの設定をオーバーライドする方法を示すためのものです。
  bootstrap.servers=localhost:9092
  
  # コンバーターは、Kafka のデータ形式と、それを Connect データに変換する方法を指定します。すべての Connect ユーザーは、Kafka からロードまたは Kafka に保存するデータの形式に基づいて、これらを設定する必要があります。
  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=true
  
  offset.storage.file.filename=/tmp/connect.offsets
  # 通常より高速にフラッシュします。これはテストやデバッグに役立ちます。
  offset.flush.interval.ms=10000
  
  # プラグイン (コネクター、コンバーター、トランスフォーメーション) のクラスローディング分離を有効にするには、ファイルシステムパスのリストをカンマ (、) で区切って設定します。リストは、以下のいずれかの組み合わせを含むトップレベルのディレクトリで構成する必要があります。
  #
  # a) プラグインとその依存関係を含む JAR を直接含むディレクトリ
  # b) プラグインとその依存関係を含む uber-jar
  # c) プラグインとその依存関係のクラスのパッケージディレクトリ構造を直接含むディレクトリ
  # 注：シンボリックリンクは、依存関係やプラグインを検出するためにたどられます。
  # 例：
  # plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
  #plugin.path=

• コネクター設定ファイルのサンプル： connect-tablestore-sink-quickstart.properties

  コネクター設定には、コネクタークラス、Tablestore 接続、データマッピングなどのパラメーターが含まれます。 詳細については、「設定」をご参照ください。

  # コネクター名を設定します。
  name=tablestore-sink
  # コネクタークラスを指定します。
  connector.class=TableStoreSinkConnector
  # タスクの最大数を設定します。
  tasks.max=1
  # データのエクスポート元となる Kafka トピックのリストを指定します。
  topics=test
  
  # 以下は Tablestore 接続パラメーターの設定です。
  # Tablestore インスタンスのエンドポイント。
  tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
  # ご自身の AccessKey ID と AccessKey Secret。
  tablestore.access.key.id =xxx
  tablestore.access.key.secret=xxx
  # Tablestore インスタンス名。
  tablestore.instance.name=xxx
  
  # 宛先の Tablestore テーブル名のフォーマット文字列。 は元のトピック名のプレースホルダーです。デフォルト値は  です。
  # 例：
  # table.name.format=kafka_ の場合、'test' トピックからのメッセージレコードは 'kafka_test' データテーブルに書き込まれます。
  table.name.format=
  
  # プライマリキーモード。デフォルト値は kafka です。
  # Tablestore データテーブルのプライマリキーは、_ (アンダースコア "_" で区切られた Kafka のトピックとパーティション) と  (パーティション内のメッセージレコードのオフセット) で構成されます。
  # primarykey.mode=
  
  # 宛先テーブルを自動的に作成します。デフォルト値は false です。
  auto.create=true

bin/connect-standalone.sh config/connect-standalone.properties config/connect-tablestore-sink-quickstart.properties

ワーカー設定には、Kafka 接続設定、シリアル化形式、オフセットのコミット頻度などのパラメーターが含まれます。 また、各コネクターの情報を保存するトピックも指定します。 これらのトピックは事前に作成しておく必要があります。 この例は、Kafka の公式ドキュメントに基づいています。 詳細については、「Kafka Connect」をご参照ください。

• offset.storage.topic： 各コネクターのオフセットを保存するために使用されるコンパクトトピック。

• config.storage.topic： コネクターとタスクの設定を保存するために使用されるコンパクトトピック。 このトピックのパーティション数は 1 に設定する必要があります。

• status.storage.topic： Kafka Connect のステータス情報を保存するために使用されるコンパクトトピック。

##
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##

# このファイルには、Kafka Connect 分散ワーカーの設定の一部が含まれています。このファイルはサンプルで使用することを目的としており、一部の設定 (特に `bootstrap.servers` やレプリケーション係数を指定する設定) は本番システムで使用されるものとは異なる場合があります。

# Kafka クラスターへの初期接続を確立するために使用するホスト/ポートのペアのリスト。
bootstrap.servers=localhost:9092

# Connect クラスターグループを形成するために使用される、クラスターの一意の名前。これはコンシューマーグループ ID と競合してはなりません。
group.id=connect-cluster

# コンバーターは、Kafka のデータ形式と、それを Connect データに変換する方法を指定します。すべての Connect ユーザーは、Kafka からロードまたは Kafka に保存するデータの形式に基づいて、これらを設定する必要があります。
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=true

# オフセットを保存するために使用するトピック。このトピックは多数のパーティションを持ち、レプリケートおよびコンパクションされる必要があります。
# Kafka Connect は必要に応じてトピックを自動的に作成しようとしますが、特定のトピック設定が必要な場合は、Kafka Connect を起動する前にいつでも手動でトピックを作成できます。
# ほとんどのユーザーは、組み込みのデフォルトのレプリケーション係数 3 を使用するか、場合によってはさらに大きな値を指定します。
# これは、使用される最大のレプリケーション係数と同じ数のブローカーが少なくとも必要であることを意味するため、この例を単一ブローカーのクラスターで実行できるように、ここではレプリケーション係数を 1 に設定しています。
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
#offset.storage.partitions=25

# コネクターとタスクの設定を保存するために使用するトピック。これは単一パーティションで、高度にレプリケートされ、コンパクションされたトピックである必要があることに注意してください。Kafka Connect は必要に応じてトピックを自動的に作成しようとしますが、特定のトピック設定が必要な場合は、Kafka Connect を起動する前にいつでも手動でトピックを作成できます。
# ほとんどのユーザーは、組み込みのデフォルトのレプリケーション係数 3 を使用するか、場合によってはさらに大きな値を指定します。
# これは、使用される最大のレプリケーション係数と同じ数のブローカーが少なくとも必要であることを意味するため、この例を単一ブローカーのクラスターで実行できるように、ここではレプリケーション係数を 1 に設定しています。
config.storage.topic=connect-configs
config.storage.replication.factor=1

# ステータスを保存するために使用するトピック。このトピックは複数のパーティションを持つことができ、レプリケートおよびコンパクションされる必要があります。
# Kafka Connect は必要に応じてトピックを自動的に作成しようとしますが、特定のトピック設定が必要な場合は、Kafka Connect を起動する前にいつでも手動でトピックを作成できます。
# ほとんどのユーザーは、組み込みのデフォルトのレプリケーション係数 3 を使用するか、場合によってはさらに大きな値を指定します。
# これは、使用される最大のレプリケーション係数と同じ数のブローカーが少なくとも必要であることを意味するため、この例を単一ブローカーのクラスターで実行できるように、ここではレプリケーション係数を 1 に設定しています。
status.storage.topic=connect-status
status.storage.replication.factor=1
#status.storage.partitions=5

# 通常より高速にフラッシュします。これはテストやデバッグに役立ちます。
offset.flush.interval.ms=10000

# これらは、REST ホストとポートの設定が存在することをユーザーに通知するために提供されます。
# REST API がリッスンするホスト名とポート。これが設定されている場合、リクエストをリッスンするために使用されるインターフェイスにバインドされます。
#rest.host.name=
#rest.port=8083

# 他のワーカーが接続するために通知されるホスト名とポート (つまり、他のサーバーからルーティング可能な URL)。
#rest.advertised.host.name=
#rest.advertised.port=

# プラグイン (コネクター、コンバーター、トランスフォーメーション) のクラスローディング分離を有効にするには、ファイルシステムパスのリストをカンマ (、) で区切って設定します。リストは、以下のいずれかの組み合わせを含むトップレベルのディレクトリで構成する必要があります。
#
# a) プラグインとその依存関係を含む JAR を直接含むディレクトリ
# b) プラグインとその依存関係を含む uber-jar
# c) プラグインとその依存関係のクラスのパッケージディレクトリ構造を直接含むディレクトリ
# 例：
# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
#plugin.path=

bin/connect-distributed.sh config/connect-distributed.properties