すべてのプロダクト
Search

このプロダクト

    ApsaraMQ for Kafka:Tablestore sink コネクタの作成

    ドキュメントセンター

    ApsaraMQ for Kafka:Tablestore sink コネクタの作成

    最終更新日:Jun 22, 2026

    このトピックでは、Tablestore sink コネクタを作成して、ApsaraMQ for Kafka インスタンスのトピックから Tablestore にデータをエクスポートする方法について説明します。

    前提条件

    注意事項

    • ApsaraMQ for Kafka インスタンスのソース トピックから Tablestore へのデータエクスポートは、同一リージョン内でのみ可能です。コネクタの制限事項については、「使用制限」をご参照ください。

    • コネクタを作成すると、ApsaraMQ for Kafka は自動的にサービスリンクロールを作成します。

      • サービスリンクロールが存在しない場合、ApsaraMQ for Kafka は自動的にロールを作成し、ApsaraMQ for Kafka が Tablestore にデータをエクスポートできるようにします。

      • サービスリンクロールが既に存在する場合、ApsaraMQ for Kafka は新しいロールを作成しません。

      サービスリンクロールについての詳細は、「サービスリンクロール」をご参照ください。

    操作手順

    このトピックでは、Tablestore sink コネクタを使用して、ApsaraMQ for Kafka インスタンスのソース トピックから Tablestore にデータをエクスポートする方法について説明します。

    1. オプション： Tablestore sink コネクタに必要なトピックとグループを作成します。

      トピックとグループをカスタマイズする必要がない場合は、このステップをスキップして、次のステップで [自動] を選択できます。

      重要

      Tablestore sink コネクタで必要な一部のトピックでは、ローカルストレージエンジンを使用する必要があります。メジャーバージョン 0.10.2 の ApsaraMQ for Kafka インスタンスでは、ローカルストレージエンジンを使用するトピックを手動で作成することはできません。これらのトピックは自動的にのみ作成できます。

      1. Tablestore sink コネクタ用のトピックの作成

      2. Tablestore sink コネクタ用のグループの作成

    2. Tablestore sink コネクタの作成とデプロイ

    3. 結果の検証

      1. テストメッセージの送信

      2. テーブル内のデータの表示

    Tablestore sink コネクタ用のトピックの作成

    ApsaraMQ for Kafka コンソールで、Tablestore sink コネクタに必要な5つのトピック (タスクオフセットトピック、タスク設定トピック、タスクステータストピック、デッドレターキュートピック、エラーデータトピック) を手動で作成します。必要なパーティション数とストレージエンジンはトピックごとに異なります。詳細については、「ソースサービスのパラメーター一覧」をご参照ください。

    1. ApsaraMQ for Kafka コンソールにログインします。

    2. [概要] ページで、[リソースの分布] セクションのリージョンを選択します。

      重要

      アプリケーションと同じリージョン、つまり ECS インスタンスがデプロイされているリージョンにトピックを作成する必要があります。トピックはリージョンをまたいで使用することはできません。たとえば、トピックが中国 (北京) リージョンで作成された場合、メッセージプロデューサーとコンシューマーも中国 (北京) リージョンの ECS インスタンス上で実行する必要があります。

    3. [インスタンスリスト] ページで、ターゲットインスタンスの名前をクリックします。

    4. 左側のナビゲーションウィンドウで、[トピック管理] をクリックします。

    5. [トピック管理] ページで、[トピックの作成] をクリックします。

    6. [トピックの作成] パネルで、トピックのプロパティを設定し、[OK] をクリックします。

      パラメーター

      説明

      例

      名前

      トピックの名前。

      説明

      Kafka では、xxx_xxx や xxx.xxx のようなトピック名は同じものと見なされます。重複した名前でトピックを作成しようとすると、エラーが報告されます。

      demo

      記述

      トピックの簡単な説明。

      demo test

      パーティションの数

      トピック内のパーティション数。

      12

      ストレージエンジン

      説明

      現在、ストレージエンジンのタイプを選択できるのは、Serverless ではない Professional Edition インスタンスのみです。他のインスタンスでは、このパラメーターはサポートされておらず、デフォルトで [クラウドストレージ] に設定されます。

      トピックメッセージのストレージエンジン。

      ApsaraMQ for Kafka は、次の2つのストレージエンジンをサポートしています。

      • [クラウドストレージ]：基盤となるレイヤーは Alibaba Cloud ディスクを使用します。このストレージエンジンは、低レイテンシ、高性能、高耐久性、高信頼性を特徴としています。分散3レプリカメカニズムを使用します。インスタンスの [仕様タイプ] が [Standard Edition (High Write)] の場合、ストレージエンジンは [クラウドストレージ] のみになります。

      • [ローカルストレージ]：ネイティブの Kafka In-Sync Replica (ISR) レプリケーションアルゴリズムと分散3レプリカメカニズムを使用します。

      クラウドストレージ

      メッセージタイプ

      トピックメッセージのタイプ。

      • [通常のメッセージ]：デフォルトでは、同じキーを持つメッセージは同じパーティションに分散され、パーティション内のメッセージは送信された順序で保存されます。クラスター内のマシンが停止した場合、メッセージの順序が乱れる可能性があります。[ストレージエンジン] を [クラウドストレージ] に設定すると、デフォルトで [通常のメッセージ] が選択されます。

      • [パーティション順位メッセージ]：デフォルトでは、同じキーを持つメッセージは同じパーティションに分散され、パーティション内のメッセージは送信された順序で保存されます。クラスター内のマシンが停止した場合でも、パーティション内でのメッセージの順序は保証されます。ただし、一部のパーティションへのメッセージ送信が失敗する可能性があります。パーティションが回復すると、通常の操作が再開されます。[ストレージエンジン] を [ローカルストレージ] に設定すると、デフォルトで [パーティション順位メッセージ] が選択されます。

      通常のメッセージ

      ログリリースポリシー

      トピックログのクリーンアップポリシー。

      [ストレージエンジン] を [ローカルストレージ] に設定する場合は、[ログリリースポリシー] を設定する必要があります。ストレージエンジンをローカルストレージに設定できるのは、Professional Edition インスタンスのみです。Standard Edition インスタンスはサポートされていません。

      ApsaraMQ for Kafka は、次の2つのクリーンアップポリシーをサポートしています。

      • [Delete]：デフォルトのメッセージクリーンアップポリシー。ディスク容量が十分な場合、メッセージは最大保持期間保持されます。ディスク容量が不十分な場合 (通常、ディスク使用率が85%を超えた場合)、サービス可用性を確保するために古いメッセージが早期に削除されます。

      • [Compact]：Kafka ログコンパクションクリーンアップポリシーを使用します。ログコンパクションポリシーは、同じキーを持つメッセージについて、常に最新の値が保持されることを保証します。これは主に、システム障害後の状態回復や、システム再起動後のキャッシュの再読み込みなどのシナリオに適用されます。たとえば、Kafka Connect や Confluent Schema Registry を使用する場合、システムの状態や設定情報を保存するために Kafka 圧縮トピックを使用する必要があります。

        重要

        圧縮トピックは、通常、Kafka Connect や Confluent Schema Registry などの特定のエコシステムコンポーネントでのみ使用されます。他のメッセージ送受信シナリオのトピックにはこのプロパティを設定しないでください。詳細については、「ApsaraMQ for Kafka デモライブラリ」をご参照ください。

      Compact

      タグ

      トピックのタグ。

      demo

      トピックが作成されると、[トピック管理] ページのリストに表示されます。

    Tablestore sink コネクタ用のグループの作成

    ApsaraMQ for Kafka コンソールで、Tablestore sink コネクタのデータ同期タスク用のグループを手動で作成できます。このグループの名前は connect-Task Name である必要があります。詳細については、「ソースサービスのパラメーター」をご参照ください。

    1. ApsaraMQ for Kafka コンソールにログインします。

    2. [概要] ページで、[リソースの分布] セクションのリージョンを選択します。

    3. [インスタンスリスト] ページで、ターゲットインスタンスの名前をクリックします。

    4. 左側のナビゲーションウィンドウで、[Group の管理] をクリックします。

    5. [Group の管理] ページで、[グループの作成] をクリックします。

    6. [グループの作成] パネルで、[Group ID] テキストボックスにグループの名前を入力し、[記述] テキストボックスに簡単な説明を入力し、グループにタグを追加してから、[OK] をクリックします。

      グループが作成されると、[Group の管理] ページのリストに表示されます。

    Tablestore sink コネクタの作成とデプロイ

    Tablestore sink コネクタを作成してデプロイし、ApsaraMQ for Kafka インスタンスから Tablestore テーブルにデータを同期します。

    1. ApsaraMQ for Kafka コンソールにログインします。

    2. [概要] ページで、[リソースの分布] セクションのリージョンを選択します。

    3. 左側のナビゲーションウィンドウで、[Connector タスクリスト] をクリックします。

    4. [Connector タスクリスト] ページで、[インスタンスの選択] ドロップダウンリストからコネクタが属するインスタンスを選択し、[Connector の作成] をクリックします。

    5. [Connector の作成] ウィザードで、次のステップを実行します。

      1. [基本情報の設定] タブで、次のパラメーターを設定し、[次へ] をクリックします。

        パラメーター

        説明

        例

        名前

        コネクタの名前。名前は次の要件を満たす必要があります：

        • 名前は最大 48 文字で、数字、小文字、ハイフン (-) のみを含めることができます。ハイフン (-) で始めることはできません。

        • 名前は ApsaraMQ for Kafka インスタンス内で一意である必要があります。

        コネクタのデータ同期タスクは、connect-task-name という名前の コンシューマーグループを使用します。この コンシューマーグループを手動で作成しない場合、システムが自動的に作成します。

        kafka-ts-sink

        インスタンス

        デフォルトでは、インスタンス名とインスタンス ID が表示されます。

        demo alikafka_post-cn-st21p8vj****

      2. [ソースサービスの設定] タブで、[データソース] を [ApsaraMQ for Kafka] に設定し、次のパラメーターを設定してから [次へ] をクリックします。

        説明

        必要なトピックとグループをすでに作成している場合は、リソース作成に [手動] を選択し、その情報を入力します。それ以外の場合は、[自動] を選択します。

        表 1. ソースサービスのパラメーター

        パラメーター

        説明

        例

        データソース Topic

        データ同期のソース トピック。

        ts-test-input

        コンシューマースレッドの同時発生数

        データソース トピックの同時コンシューマースレッド数。デフォルト値は 6 です。有効な値：

        • 1

        • 2

        • 3

        • 6

        • 12

        6

        消費の開始位置

        メッセージ消費の開始位置。有効な値：

        • [一番古いオフセット]：最小オフセットからメッセージを消費します。

        • [一番新しいオフセット]：最大オフセットからメッセージを消費します。

        一番古いオフセット

        VPC ID

        データ同期タスクが実行される VPC の ID。このパラメーターは、[実行環境の設定] をクリックすると表示されます。このパラメーターは、ApsaraMQ for Kafka インスタンスの VPC にデフォルト設定されており、設定は不要です。

        vpc-bp1xpdnd3l***

        VSwitch ID

        データ同期タスクが実行される vSwitch の ID。このパラメーターは、[実行環境の設定] をクリックすると表示されます。vSwitch は、ApsaraMQ for Kafka インスタンスと同じ VPC にある必要があります。このパラメーターは、ApsaraMQ for Kafka インスタンスがデプロイされたときに指定された vSwitch にデフォルト設定されています。

        vsw-bp1d2jgg81***

        失敗の処理

        トピックパーティションでのメッセージ送信失敗時の処理ポリシー。このパラメーターは、[実行環境の設定] をクリックすると表示されます。有効な値：

        • [サブスクリプションの継続]：エラーが発生したトピックパーティションへのサブスクリプションを続行し、エラーログを出力します。

        • [サブスクリプションの停止]：エラーが発生したトピックパーティションへのサブスクリプションを停止し、エラーログを出力します。

        説明

        • ログの表示方法については、「コネクタの管理」をご参照ください。

        • エラーコードに基づいた解決策の検索方法については、「エラーコード」をご参照ください。

        サブスクリプションの継続

        リソースの作成方法

        必須のトピックとグループを作成するためのメソッド。このパラメーターは、実行環境の設定 をクリックすると表示されます。

        • 自動作成

        • 手動で作成します

        自動作成

        Connector コンシューマーグループ

        データ同期タスクで使用されるコンシューマーグループ。このパラメーターは、[実行環境の設定] をクリックすると表示されます。グループの名前は connect-task-name の形式である必要があります。

        connect-cluster-kafka-ots-sink

        タスクサイトの Topic

        コンシューマーオフセットを保存するトピック。このパラメーターは、[実行環境の設定] をクリックすると表示されます。

        • トピック：トピック名は connect-offset で始まる必要があります。

        • パーティション：トピックのパーティション数は 1 より大きい必要があります。

        • ストレージエンジン：トピックのストレージエンジンはローカルストレージである必要があります。

        • cleanup.policy：トピックのログクリーンアップポリシーは compact である必要があります。

        connect-offset-kafka-ots-sink

        タスク設定の Topic

        タスク設定を保存するトピック。このパラメーターは、[実行環境の設定] をクリックすると表示されます。

        • トピック：トピック名は connect-config で始まる必要があります。

        • パーティション：トピックのパーティション数は 1 である必要があります。

        • ストレージエンジン：トピックのストレージエンジンはローカルストレージである必要があります。

        • cleanup.policy：トピックのログクリーンアップポリシーは compact である必要があります。

        connect-config-kafka-ots-sink

        [タスクステータスの Topic]

        タスクステータスを保存するトピック。このパラメーターは、[実行環境の設定] をクリックすると表示されます。

        • トピック：トピック名は connect-status で始まる必要があります。

        • パーティション：推奨されるパーティション数は 6 です。

        • ストレージエンジン：トピックのストレージエンジンはローカルストレージである必要があります。

        • cleanup.policy：トピックのログクリーンアップポリシーは compact である必要があります。

        connect-status-kafka-ots-sink

        デッドレターキューの Topic

        Kafka Connect フレームワークからのエラーデータを保存するために使用されるトピック。このパラメーターは、[実行環境の設定] をクリックすると表示されます。トピックリソースを節約するために、デッドレターキューと [エラーデータトピック] に同じトピックを使用できます。

        • トピック：トピック名は connect-error で始まる必要があります。

        • パーティション：推奨されるパーティション数は 6 です。

        • ストレージエンジン：トピックのストレージエンジンは、ローカルストレージまたはクラウドストレージのいずれかです。

        connect-error-kafka-ots-sink

        例外データ Topic

        シンクからのエラーデータを保存するために使用されるトピック。このパラメーターは、[実行環境の設定] をクリックすると表示されます。トピックリソースを節約するために、[デッドレターキュートピック] とエラーデータトピックに同じトピックを使用できます。

        • トピック：トピック名は connect-error で始まる必要があります。

        • パーティション：推奨されるパーティション数は 6 です。

        • ストレージエンジン：トピックのストレージエンジンは、ローカルストレージまたはクラウドストレージのいずれかです。

        connect-error-kafka-ots-sink

      3. [ターゲットサービスの設定] タブで、[宛先サービス] を [Tablestore] に設定し、次のパラメーターを設定してから [作成] をクリックします。

        パラメーター

        説明

        例

        インスタンス名

        Tablestore インスタンスの名前。

        k00eny67****

        [宛先テーブルの自動作成]

        Tablestore にテーブルを自動的に作成するかどうかを指定します。

        • [はい]：設定されたテーブル名に基づいて、同期データを保存するためのテーブルが Tablestore に自動的に作成されます。

        • [いいえ]：既存のテーブルを使用して同期データを保存します。

        はい

        宛先テーブル名

        同期データを保存するテーブルの名前。[宛先テーブルの自動作成] を [いいえ] に設定した場合、テーブル名は Tablestore インスタンス内の既存のテーブルの名前と同じである必要があります。

        kafka_table

        Tablestore

        同期データを保存するテーブルのタイプ。

        • ワイドカラムモデル

        • 時系列モデル

        ワイドカラムモデル

        メッセージキー形式

        メッセージキーの入力フォーマット。有効な値は String と JSON です。デフォルト値は JSON です。このパラメーターは、[Tablestore] が [ワイドカラムモデル] に設定されている場合にのみ表示されます。

        • String：メッセージキーは文字列として直接解析されます。

        • JSON：メッセージキーは JSON 形式である必要があります。

        String

        メッセージ値の形式

        メッセージ値の入力フォーマット。有効な値は String と JSON です。デフォルト値は JSON です。このパラメーターは、[Tablestore] が [ワイドカラムモデル] に設定されている場合にのみ表示されます。

        • String：メッセージ値は文字列として直接解析されます。

        • JSON：メッセージ値は JSON 形式である必要があります。

        String

        JSON メッセージフィールド変換

        JSON メッセージ内のフィールドを処理する方法。このパラメーターは、[メッセージキー形式] または [メッセージ値の形式] を JSON に設定した場合に表示されます。有効な値：

        • すべて文字列として書き込む：すべてのフィールドを Tablestore の String 型に変換します。

        • フィールドタイプを自動的に識別：JSON メッセージ本文の String および Boolean フィールドを、Tablestore の対応する String および Boolean 型に変換します。JSON メッセージ本文の Integer および Float データ型は、Tablestore の Double 型に変換されます。

        すべて文字列として書き込む

        主キーモード

        プライマリキーモードを指定します。テーブルのプライマリキーを、座標 (トピック、パーティション、オフセット)、キー、値など、ApsaraMQ for Kafka メッセージレコードのさまざまな部分から抽出できます。このパラメーターは、[Tablestore] が [ワイドカラムモデル] に設定されている場合にのみ表示されます。デフォルト値は kafka です。

        • kafka：<connect_topic>_<connect_partition> と <connect_offset> をデータテーブルのプライマリキーとして使用します。

        • record_key：レコードキーのフィールドをデータテーブルのプライマリキーとして使用します。

        • record_value：レコード値のフィールドをデータテーブルのプライマリキーとして使用します。

        kafka

        プライマリキー列名の設定

        データテーブルのプライマリキー列名とそれに対応するデータ型。String と Integer のデータ型がサポートされています。これは、設定された列名に一致するレコードキーまたは値のフィールドが、データテーブルのプライマリキーとして使用されることを意味します。

        このパラメーターは、[メッセージキー形式] を JSON に、[主キーモード] を record_key に設定した場合、または [メッセージ値の形式] を JSON に、[主キーモード] を record_value に設定した場合に表示されます。

        [追加] をクリックして列名を追加します。最大4つの列名を設定できます。

        なし

        書き込みモード

        書き込みモードを指定します。有効な値は put と update です。デフォルト値は put です。このパラメーターは、[Tablestore] が [ワイドカラムモデル] に設定されている場合にのみ表示されます。

        • put：既存のデータを上書きします。

        • update：既存のデータを更新します。

        put

        削除モード

        ApsaraMQ for Kafka メッセージレコードに null 値が含まれている場合、行または属性列を削除するかどうかを選択できます。このパラメーターは、[主キーモード] が record_key に設定されている場合に表示されます。有効な値：

        • none：デフォルト値。削除は許可されません。

        • row：行の削除を許可します。

        • column：属性列の削除を許可します。

        • row_and_column：行と属性列の削除を許可します。

        削除の動作は書き込みモードによって異なります：

        • [書き込みモード] が put の場合、値に null フィールドが含まれている場合でも、どの削除モードでも Tablestore データテーブルへの上書き書き込みが行われます。

        • [書き込みモード] が update で、削除モードが none または row の場合、すべての値フィールドが null のレコードはダーティデータとして扱われます。一部の値フィールドのみが null の場合、コネクタは null フィールドを無視し、null でないフィールドを Tablestore データテーブルに書き込みます。削除モードが column または row_and_column の場合、コネクタは null フィールドの行と属性列を削除してから、データを Tablestore データテーブルに書き込みます。

        なし

        メジャー名フィールド

        このフィールドは、Tablestore 時系列モデルのメジャー名フィールド (_m_name) にマッピングされます。メジャー名は、温度や速度など、時系列データの物理量またはモニタリングメトリックの名前を指定します。このフィールドは空にできません。このパラメーターは、[Tablestore] に時系列モデルを選択した場合に表示されます。

        measurement

        データソースフィールド

        このフィールドを Tablestore 時系列モデルのデータソースフィールド (_data_source) にマッピングします。このフィールドは、マシン名やデバイス ID など、時系列データのソースの識別子として機能し、空にすることができます。このパラメーターは、[Tablestore] に時系列モデルを選択した場合に表示されます。

        source

        ラベルフィールド

        1つ以上のフィールドを Tablestore 時系列モデルのタグフィールド (_tags) として使用します。各タグは文字列のキーと値のペアです。キーは設定されたフィールドの名前で、値はフィールドの内容です。タグは時系列メタデータの一部です。時系列は、メトリック名、データソース、およびタグの組み合わせによって一意に識別されます。タグは空にすることができます。このパラメーターは、[Tablestore] が時系列モデルとして選択されている場合に表示されます。

        tag1, tag2

        タイムスタンプフィールド

        このフィールドを Tablestore 時系列モデルのタイムスタンプフィールド (_time) にマッピングします。これは、物理量が生成された時間など、この時系列データの行の時点を表します。データが Tablestore に書き込まれるとき、タイムスタンプフィールドは書き込みと保存のためにマイクロ秒に変換されます。このパラメーターは、[Tablestore] に時系列モデルを選択した場合に表示されます。

        time

        タイムスタンプ単位

        実際のタイムスタンプフィールドに基づいてこのパラメーターを設定します。このパラメーターは、[Tablestore] が [時系列モデル] に設定されている場合にのみ表示されます。有効な値：

        • SECONDS

        • MILLISECONDS

        • MICROSECONDS

        • NANOSECONDS

        MILLISECONDS

        [すべての非主キーフィールドのマッピングの設定を行うかどうか]

        すべての非プライマリキーフィールドをデータフィールドとしてマッピングするかどうかを指定します。非プライマリキーフィールドとは、メトリック名、データソース、タグ、またはタイムスタンプとしてまだマッピングされていないフィールドです。このパラメーターは、[Tablestore] が [時系列モデル] に設定されている場合にのみ表示されます。有効な値：

        • [はい]：フィールドは自動的にマッピングされ、データ型が決定されます。数値型はすべて Double 型に変換されます。

        • [いいえ]：マッピングするフィールドと型を指定する必要があります。

        はい

        [すべての非主キーフィールドのマッピングの設定]

        時系列テーブルの非プライマリキーフィールド名に対応するフィールドタイプ。Double、Integer、String、Binary、Boolean のデータ型がサポートされています。このパラメーターは、[すべての非主キーフィールドのマッピングの設定を行うかどうか] を [いいえ] に設定した場合に表示されます。

        String

        コネクタが作成されると、[Connector タスクリスト] ページで表示できます。

    6. コネクタが作成された後、[Connector タスクリスト] ページでそれを見つけ、[操作] 列の [デプロイ] をクリックします。

    7. [OK] をクリックします。

    テストメッセージの送信

    Tablestore sink コネクタをデプロイした後、ApsaraMQ for Kafka のソース トピックにメッセージを送信して、データが Tablestore に同期されることを確認できます。

    1. [Connector タスクリスト] ページで、ターゲットコネクタを見つけ、[操作] 列の [テスト] をクリックします。

    2. [メッセージの送信] パネルで、テストメッセージを送信します。

      • [送信方法] を [コンソール] に設定します。

        1. [メッセージキー] テキストボックスに、メッセージのキーを入力します。たとえば、demo です。

        2. [メッセージの内容] テキストボックスに、テストメッセージの内容を入力します。たとえば、{"key": "test"} です。

        3. [指定されたパーティションに送信] を設定して、メッセージを特定のパーティションに送信するかどうかを指定します。

          • [はい] をクリックし、[パーティション ID] テキストボックスにパーティション ID を入力します。たとえば、0 です。パーティション ID をクエリするには、「パーティションステータスの表示」をご参照ください。

          • [いいえ] をクリックして、パーティションを指定しません。

      • [送信方法] を [Docker] に設定します。[Docker コンテナーを実行してサンプルメッセージを生成する] セクションの Docker コマンドを実行してメッセージを送信します。

      • [送信方法] を [SDK] に設定します。必要な言語またはフレームワークの SDK と接続タイプを選択してメッセージを送信します。

    テーブルデータの表示

    ApsaraMQ for Kafka のデータソース トピックにメッセージを送信した後、Tablestore テーブルのデータを表示して、メッセージが受信されたことを確認します。

    1. Tablestore コンソールにログインします。

    2. [概要] ページで、インスタンス名をクリックするか、[操作] 列の [Instances] をクリックします。

    3. [インスタンスの詳細] タブで、[テーブル] セクションのターゲットテーブルを見つけます。

    4. テーブル名をクリックします。[テーブル管理] ページで、[データ管理] タブをクリックしてデータを表示します。