すべてのプロダクト
Search

このプロダクト

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

    ドキュメントセンター

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

    最終更新日:Jun 22, 2026

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

    前提条件

    • Tablestore が有効化され、インスタンスが作成されています。 詳細については、「Tablestore を有効化してインスタンスを作成する」をご参照ください。

    • ApsaraMQ for Kafka インスタンスを購入して有効化し、トピックを作成済みであること。詳細な手順については、「ApsaraMQ for Kafka インスタンスの購入」および「リソースの作成」をご参照ください。

    • Tablestore sink コネクタタスクによって生成されるサービスリンクロールには、AliyunOTSFullAccess ポリシーが必要です。このポリシーを手動でアタッチして、ロールに Tablestore の管理権限を付与してください。詳細については、「RAM ロールへの権限付与」をご参照ください。

    ステップ 1: Tablestore テーブルの作成

    ApsaraMQ for Kafka からストリーミングされたデータを格納するための Tablestore テーブルを作成します。詳細については、「手順」をご参照ください。

    このトピックでは、`ots-sink` という名前のインスタンスと `ots_sink_table` という名前のデータテーブルを例として使用します。テーブルを作成する際に、3 つのプライマリキーを定義します。STRING 型の topic (パーティションキーとして設定)、INTEGER 型の partition、および INTEGER 型の offset です。

    ステップ 2: Tablestore sink コネクタの作成と開始

    1. ApsaraMQ for Kafka コンソールにログオンします。概要 ページで、リソースの分布 セクションでリージョンを選択します。

    2. 左側のナビゲーションウィンドウで、Connector エコシステムの統合 > タスクリストを選択します。

    3. タスクリスト ページで、タスクの作成 をクリックします。

    4. タスクの作成 ページで、Task Name と Description を設定し、次のパラメーターを設定して、保存 をクリックします。

      • タスクの作成

        1. Source (ソース) ステップで、データプロバイダー に Message Queue for Apache Kafka を選択します。次のパラメーターを設定し、[次へ] をクリックします。

          パラメーター

          説明

          例

          リージョン

          ソースの Message Queue for Apache Kafka インスタンスのリージョン。

          中国 (北京)

          Kafka インスタンス

          ソースの Message Queue for Apache Kafka インスタンス。

          alikafka_post-cn-jte3****

          トピック

          Message Queue for Apache Kafka メッセージを生成するトピックを選択します。

          demo-topic

          グループ ID

          ソースインスタンスのコンシューマーグループ。

          • クイック作成: 推奨オプションです。GID_EVENTBRIDGE_xxx 形式のグループ ID が自動的に作成されます。

          • 既存のものを使用: 独立したグループ ID を選択します。既存のメッセージ消費を妨げないように、他のサービスと共有されているグループ ID は使用しないでください。

          クイック作成

          コンシューマーオフセット

          メッセージ消費を開始するオフセット。

          • 最新のオフセット (latest)

          • 最も早いオフセット (earliest)

          最新のオフセット (latest) 

          ネットワーク設定

          メッセージをルーティングするためのネットワークタイプ。

          • ベーシックネットワーク

          • 自己管理型インターネット

          ベーシックネットワーク

          VPC

          [ネットワーク設定] が [自己管理型インターネット] に設定されている場合にのみ必須です。

          vpc-bp17fapfdj0dwzjkd****

          vSwitch

          [ネットワーク設定] が [自己管理型インターネット] に設定されている場合にのみ必須です。

          vsw-bp1gbjhj53hdjdkg****

          セキュリティグループ

          [ネットワーク設定] が [自己管理型インターネット] に設定されている場合にのみ必須です。

          alikafka_pre-cn-7mz2****

          データ形式

          メッセージ本文のエンコード形式です。バイナリ転送をサポートするデータソースの場合は、次のいずれかの形式を選択します。特定のエンコード要件がない場合は、Json を推奨します。

          • Json: バイナリデータを UTF-8 を使用して JSON オブジェクトとしてペイロードにエンコードします。

          • Text: バイナリデータを UTF-8 文字列としてペイロードにエンコードします。これがデフォルトの形式です。

          • Binary: バイナリデータを Base64 エンコードされた文字列としてペイロードにエンコードします。

          Json

          一括プッシュの件数

          [高度な設定] パラメーターです。バッチごとに送信するメッセージの最大数。蓄積されたメッセージ数が指定された値に達した場合にのみリクエストが送信されます。有効値: 1～10,000。

          100

          バッチプッシュ間隔 (単位：秒)

          [高度な設定] パラメーターです。関数を呼び出す間隔。システムはメッセージを集約し、各間隔で sink に送信します。有効値: 0～15。値 0 は、遅延なくメッセージがすぐに配信されることを指定します。

          3

        2. Filtering (フィルタリング) ステップで、パターン内容 を設定してイベントをフィルターします。詳細については、「イベントパターン」をご参照ください。

        3. Transform (変換) ステップで、分割、マッピング、エンリッチメント、動的ルートなどの操作を実行するようにデータ変換を設定します。詳細については、「Function Compute を使用してメッセージデータをクリーンアップする」をご参照ください。

        4. Sink (ターゲット)ステップでは、サービスタイプを Tablestore に設定し、以下のパラメーターを設定します。

          パラメーター

          説明

          例

          インスタンス名

          送信先の Tablestore インスタンスの名前。

          ots-sink

          ターゲットテーブル

          送信先の Tablestore データテーブルの名前。

          ots_sink_table

          主キー

          Tablestore のプライマリキー値と属性列を生成するには、JsonPath 構文を使用して各属性列の内容の抽出ルールを定義する必要があります。Source (ソース)のデータ形式がJSONに設定されている場合、ApsaraMQ for Kafkaからの出力データは次の形式になります。

          {
    "data": {
        "topic": "demo-topic",
        "partition": 0,
        "offset": 2,
        "timestamp": 1739756629123,
        "headers": {
            "headers": [],
            "isReadOnly": false
        },
        "key":"ots-sink-k1",
        "value": "ots-sink-v1"
    },
    "id": "7702ca16-f944-4b08-***-***-0-2",
    "source": "acs:alikafka",
    "specversion": "1.0",
    "type": "alikafka:Topic:Message",
    "datacontenttype": "application/json; charset=utf-8",
    "time": "2025-02-17T01:43:49.123Z",
    "subject": "acs:alikafka:alikafka_serverless-cn-lf6418u6701:topic:demo-topic",
    "aliyunaccountid": "1******6789"
}

          たとえば、topic という名前のプライマリキーの場合、値の抽出ルールを $.data.topic に設定します。

          属性列

          たとえば、key という名前の属性列の場合、値の抽出ルールを $.data.key に設定します。

          操作モード

          Tablestore にデータを書き込むメソッド。

          • put: 同じプライマリキーを持つレコードが既に存在する場合、新しいデータが既存のデータを上書きします。

          • update: 2 つのデータレコードが同じプライマリキーを持つ場合、増分列が行に書き込まれ、既存の列は削除されません。

          • delete: 対応するプライマリキーのデータを削除します。

          put

          ネットワーク設定

          • VPC: VPC を使用して Kafka メッセージを Tablestore に配信します。

          • Public Network: パブリックネットワーク経由で Kafka メッセージを Tablestore に配信します。

          VPC

          VPC

          VPC ID を選択します。このパラメーターは、[ネットワーク設定] が VPC の場合にのみ必須です。

          vpc-bp17fapfdj0dwzjkd****

          vSwitch

          vSwitch ID を選択します。このパラメーターは、[ネットワーク設定] が VPC に設定されている場合にのみ必須です。

          vsw-bp1gbjhj53hdjdkg****

          セキュリティグループ

          セキュリティグループを選択します。このパラメーターは、[ネットワーク構成] が VPC の場合にのみ必須です。

          test_group

      • タスクプロパティ

        失敗したイベント配信のリトライポリシーとエラー処理のメソッドを設定します。詳細については、「リトライポリシーとデッドレターキュー」をご参照ください。

    5. タスクリスト ページに戻ります。目的のタスクを見つけ、操作する 列の 有効化する をクリックします。

    6. ヒント ダイアログボックスで情報を読み取り、OK をクリックします。

      タスクを有効にすると、30～60 秒以内に開始されます。タスクリスト ページの Status 列で、そのステータスをモニターできます。

    ステップ 3: Tablestore sink コネクタのテスト

    1. タスクリスト ページで、対象タスクの イベントソース 列でソース Topic をクリックします。

    2. トピック詳細ページで、[テストメッセージの送信] をクリックします。

    3. メッセージ送受信のクイック体験 パネルでメッセージを設定し、OK をクリックします。

      [コンソール] タブで、[メッセージキー] を ots-sink-k1 に、[メッセージ内容] を ots-sink-v1 に、[指定したパーティションに送信] を [いいえ] に設定します。

    4. タスクリスト ページに戻り、タスクの イベントターゲット 列で、送信先テーブル名をクリックします。

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

      データテーブルには、topic (プライマリキー)、partition (プライマリキー)、offset (プライマリキー)、key、および value 列が含まれます。partition=2、offset=10、key=ots-sink-k1、value=ots-sink-v1 などの値を持つレコードが確認できれば、Kafka メッセージが Tablestore に書き込まれたことになります。