時系列テーブルの Kafka を設定する - Tablestore - Alibaba Cloud ドキュメントセンター

kafka-connect-tablestore パッケージを使用して、Kafka から Tablestore の時系列テーブルにデータを書き込みます。このトピックでは、このデータシンクの設定方法について説明します。

前提条件

Apache Kafkaがインストールされ、有効になっており、ZooKeeperが有効になっていること。詳細は、Kafkaドキュメントをご参照ください。
Tablestore サービスを有効化し、インスタンスと時系列テーブルを作成していること。詳細については、「TimeSeries モデルを使用する」をご参照ください。

説明
auto.create パラメータを true に設定することで、Tablestore シンクコネクターに宛先の時系列テーブルを自動的に作成させることもできます。
AccessKeyペアを取得していること。詳細は、AccessKeyペアの取得をご参照ください。

背景情報

Tablestore は時系列データを保存および分析します。詳細については、「TimeSeries モデルの概要」をご参照ください。

手順1: Tablestore Sink Connectorをデプロイする

次のいずれかの方法でTablestore Sink Connectorパッケージを取得します。
- GitHubのTablestore Sink Connectorソースコードからソースコードをダウンロードし、ソースコードをコンパイルします。
  1. Gitツールを使用して、次のコマンドを実行してTablestore Sink Connectorのソースコードをダウンロードします。
```
git clone https://github.com/aliyun/kafka-connect-tablestore.git
```
  2. ダウンロードしたソースコードが保存されているディレクトリに移動し、Mavenを使用して次のコマンドを実行してソースコードをパッケージ化します。
```
mvn clean package -DskipTests
```
    コンパイルが完了すると、生成されたパッケージはtargetディレクトリに保存されます。kafka-connect-tablestore-1.0.jarパッケージを例として使用します。
- コンパイル済みのkafka-connect-tablestoreパッケージをダウンロードします。
各ノードの$KAFKA_HOME/libsディレクトリにパッケージをコピーします。

ステップ 2： Tablestore シンクコネクターの起動

Tablestore Sink Connectorは、スタンドアロンモードまたは分散モードで動作できます。ビジネス要件に基づいてモードを選択できます。

時系列データを書き込む場合、Kafka からのメッセージレコードは JSON 形式である必要があります。そのため、Tablestore シンクコネクターを起動する際は JsonConverter を使用する必要があります。スキーマを抽出したり、キーを提供したりする必要はありません。connect-standalone.properties ファイルと connect-distributed.properties ファイルで、次のパラメータを以下のように設定します。

説明

キーを提供する場合は、キーの形式に基づいて key.converter パラメータと key.converter.schemas.enable パラメータを設定してください。

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

このセクションでは、スタンドアロンモードの例を示します。分散モードの設定は、標準データテーブルへのデータ同期と同様です。connect-distributed.properties ワーカー設定ファイルで値コンバーターの設定を更新し、connect-tablestore-sink-quickstart.json コネクター設定ファイルで時系列設定を調整する必要があります。詳細については、「ステップ 2: Tablestore シンクコネクターの起動」の分散モードの設定手順をご参照ください。

スタンドアロンモードを設定するには、次の手順を実行します。

ご使用の環境に合わせて、ワーカー設定ファイル connect-standalone.properties とコネクター設定ファイル connect-tablestore-sink-quickstart.properties を変更します。

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

ワーカー構成ファイルには、構成項目が含まれています。これらの項目には、Kafka接続パラメーター、シリアル化形式、およびオフセットがコミットされる頻度が含まれます。次のサンプルコードは、Apache 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=false
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
# シンク対象のトピックのカンマ区切りリスト
topics=test
# 以下は Tablestore 接続パラメータです。
# Tablestore インスタンスのエンドポイント
tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
# 認証モード
tablestore.auth.mode=aksk
# AccessKey ID と AccessKey シークレット。「aksk」 認証の場合に必要です。
tablestore.access.key.id=xxx
tablestore.access.key.secret=xxx
# Tablestore インスタンスの名前
tablestore.instance.name=xxx
## STS 認証設定。STS を使用する場合は、次のパラメータが必要です。 
## また、ACCESS_ID と ACCESS_KEY を環境変数として設定する必要があります。
#sts.endpoint=
#region=
#account.id=
#role.name=
# 宛先テーブル名のフォーマット文字列。 プレースホルダーを含めることができます。
# topics.assign.tables 設定は、この設定よりも優先されます。
# たとえば、table.name.format が kafka_ に設定されており、Kafka トピック名が 'test' の場合、データは 'kafka_test' という名前の Tablestore テーブルにマッピングされます。
table.name.format=<topic>
# トピックとテーブルのマッピング。形式は ":" です。トピックとテーブル名をコロン (:) で区切り、複数のマッピングをカンマ (,) で区切ります。
# これが指定されていない場合は、table.name.format 設定が使用されます。
# topics.assign.tables=test:test_kafka
# 宛先テーブルを自動的に作成するかどうかを指定します。デフォルト： false
auto.create=true
# 以下はダーティデータの処理設定です。
# Kafka レコードの解析または時系列テーブルへの書き込み時にエラーが発生する可能性があります。これらの設定を使用してエラーを管理します。
# フォールトトレランスレベルを指定します。有効な値： none、all。デフォルト： none
# none: エラーが発生すると、コネクタータスクは直ちに失敗します。
# all: エラーの原因となるレコードをスキップし、ログに記録します。
runtime.error.tolerance=none
# ダーティデータの処理モード。有効な値： ignore、kafka、tablestore。デフォルト： ignore
# ignore: すべてのエラーを無視します。
# kafka: 問題のあるレコードとエラー情報を別の Kafka トピックに保存します。
# tablestore: 問題のあるレコードとエラー情報を別の Tablestore データテーブルに保存します。
runtime.error.mode=ignore
# runtime.error.mode が 'kafka' に設定されている場合は、Kafka クラスターのアドレスとトピック名を設定します。
# runtime.error.bootstrap.servers=localhost:9092
# runtime.error.topic.name=errors
# runtime.error.mode が 'tablestore' に設定されている場合は、Tablestore データテーブルの名前を設定します。
# runtime.error.table.name=errors
## 以下は時系列テーブル固有の設定です。
# コネクターのモード。デフォルト： 通常
tablestore.mode=timeseries
# 時系列プライマリキーフィールドのマッピング
tablestore.timeseries.test.measurement=m
tablestore.timeseries.test.dataSource=d
tablestore.timeseries.test.tags=region,level
# 時系列時刻フィールドのマッピング
tablestore.timeseries.test.time=timestamp
tablestore.timeseries.test.time.unit=MILLISECONDS
# フィールド名を小文字に変換するかどうかを指定します。デフォルト： true。TimeSeries モデルでは列名に大文字を使用できません。これを false に設定し、列名に大文字が含まれている場合、書き込みエラーが発生します。
tablestore.timeseries.toLowerCase=true
# プライマリキーおよびタイムスタンプフィールド以外のすべてのフィールドを、時系列テーブルのフィールドとして保存するかどうかを指定します。デフォルト： true。false の場合、tablestore.timeseries.test.field.name で指定されたフィールドのみが保存されます。
tablestore.timeseries.mapAll=true
# フィールド名。カンマ (,) で区切ります。
tablestore.timeseries.test.field.name=cpu
# フィールドのデータ型。有効な値: double、integer、string、binary、boolean
# 複数のフィールドが指定されている場合、型はフィールド名と 1 対 1 で対応している必要があります。複数の型はカンマ (,) で区切ります。
tablestore.timeseries.test.field.type=double

$KAFKA_HOME ディレクトリに移動し、次のコマンドを実行してスタンドアロンモードを有効にします。
```
bin/connect-standalone.sh config/connect-standalone.properties config/connect-tablestore-sink-quickstart.properties
```

ステップ 3：新しいレコードの生成

$KAFKA_HOME ディレクトリに移動し、次のコマンドを実行してコンソールプロデューサークライアントを起動します。

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

次の表は、コンソールプロデューサークライアントを起動するために構成する必要があるパラメーターを示しています。

パラメーター	例	説明
--broker-list	localhost:9092	Kafka クラスター内のブローカーのアドレスとポート。
--topic	test	トピックの名前。デフォルトでは、Tablestore Sink Connector を起動するとトピックが自動的に作成されます。トピックを手動で作成することもできます。

test トピックに新しいメッセージを書き込みます。
重要
時系列テーブルにデータをインポートするには、JSON 形式でトピックにデータを書き込む必要があります。
```
{"m":"cpu","d":"127.0.0.1","region":"shanghai","level":1,"timestamp":1638868699090,"io":5.5,"cpu":3.5}
```
Tablestore コンソールにログインして、データを確認します。

[時系列テーブル管理] ページで、[データ管理] タブを選択します。メジャーメント名が cpu、データソースが 127.0.0.1、タグが level=1 および region=shanghai の時系列を見つけます。[データをクエリ] をクリックします。クエリは、時刻が 2021-12-07 17:18:19、cpu が 3.5、io が 5.5 のレコードを返します。