すべてのプロダクト
Search

このプロダクト

    ApsaraMQ for Kafka:AnalyticDB シンクコネクタの作成

    ドキュメントセンター

    ApsaraMQ for Kafka:AnalyticDB シンクコネクタの作成

    最終更新日:Jun 21, 2026

    このトピックでは、AnalyticDB シンクコネクタを作成する方法について説明します。このコネクタを使用して、ApsaraMQ for Kafka インスタンスのソース トピックから AnalyticDB データベースのテーブルにデータをストリーミングできます。

    前提条件

    詳細については、「前提条件」をご参照ください。

    ステップ 1：宛先リソースの作成

    AnalyticDB for MySQL または AnalyticDB for PostgreSQL のリソースを作成します。

    この例では、adb_sink_database という名前の AnalyticDB for MySQL データベースと、adb_sink_table という名前のテーブルを使用します。

    ステップ 2：AnalyticDB シンクコネクタの作成と有効化

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

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

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

      • タスクの設定

        1. Source (ソース) 設定ウィザードで、データプロバイダー を [ApsaraMQ For Kafka] に設定します。 次のパラメーターを設定し、[次へ] をクリックします。

          パラメーター

          説明

          例

          リージョン

          ソース ApsaraMQ for Kafka インスタンスが配置されているリージョンを選択します。

          中国 (北京)

          ApsaraMQ For Kafka インスタンス

          ApsaraMQ for Kafka メッセージを生成するソースインスタンスを選択します。

          alikafka_post-cn-jte3****

          Topic

          ApsaraMQ for Kafka メッセージを生成するための Topic を選択します。

          demo-topic

          グループ ID

          ソースインスタンスのコンシューマーグループの名前を選択します。

          • [クイック作成]: 推奨。 システムは、GID_EVENTBRIDGE_xxx という名前のグループ ID を自動的に作成します。

          • [既存のグループを使用]: 使用中でないグループ ID を選択します。 すでに使用中のグループ ID を使用すると、既存のメッセージのパブリッシュとサブスクリプションに影響します。

          クイック作成

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

          メッセージが消費されるオフセット。 有効な値:

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

          • [最も古いオフセット (earliest)]

          最新のオフセット (latest)

          ネットワーク構成

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

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

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

          クラシックネットワーク

          VPC

          VPC の ID。 このパラメーターは、[ネットワーク構成] が [自己管理インターネット] に設定されている場合に必要です。

          vpc-bp17fapfdj0dwzjkd****

          VSwitch

          ApsaraMQ for Kafka インスタンスが属する vSwitch の ID。 このパラメーターは、[ネットワーク構成] パラメーターを [自己管理インターネット] に設定した場合にのみ必要です。

          vsw-bp1gbjhj53hdjdkg****

          セキュリティグループ

          セキュリティグループを選択します。 このパラメーターは、[ネットワーク構成] が [自己管理インターネット] に設定されている場合に必要です。

          alikafka_pre-cn-7mz2****

          データ形式

          データ形式の設定では、データソースからのバイナリデータを特定の形式にエンコードします。 複数のデータ形式がサポートされています。 特別なエンコーディング要件がない場合は、形式を Json に設定します。

          • [Json]: バイナリデータは、UTF-8 エンコーディングに基づいて JSON 形式のデータにエンコードされ、ペイロードに格納されます。

          • [Text]: バイナリデータは、UTF-8 エンコーディングに基づいて文字列にエンコードされ、ペイロードに格納されます。 これがデフォルトの形式です。

          • [Binary]: バイナリデータは、Base64 エンコーディングに基づいて文字列にエンコードされ、ペイロードに格納されます。

          Json

          一括プッシュの件数

          [高度な設定] パラメーター: 各関数呼び出しで送信できるメッセージの最大数。 リクエストは、バックログ内のメッセージ数が指定された値に達した場合にのみ送信されます。 有効な値: 1～10,000。

          100

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

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

          3

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

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

        4. Sink (ターゲット) ステップで、サービスタイプ を クラウドネイティブデータウェアハウス AnalyticDB に設定し、次のパラメーターを設定してから、保存 をクリックします。

          パラメーター

          説明

          例

          インスタンスタイプ

          宛先インスタンスのデータベースタイプを選択します。

          • [MySQL Edition]

          • [PostgreSQL Edition]

          MySQL バージョン

          AnalyticDB インスタンス ID

          宛先インスタンスを選択します。

          am-bp1abcde12345****

          データベース名

          宛先データベースを選択します。

          adb_sink_database

          テーブル名

          宛先データテーブルを選択します。

          adb_sink_table

          [データマッピング]

          JSONPath 式を使用してデータ抽出ルールを定義します。Source (ソース) ステップで データ形式 が [Json] に設定されている場合、ApsaraMQ for Kafka からストリーミングされたデータは、次のように CloudEvents 構造でラップされます。

          {
    "data": {
        "topic": "demo-topic",
        "partition": 0,
        "offset": 2,
        "timestamp": 1739756629123,
        "headers": {
            "headers": [],
            "isReadOnly": false
        },
        "key":"adb-sink-k1",
        "value": {
            "userid":"xiaoming",
            "source":"shanghai"
        }
    },
    "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"
}

          JSONPath 式を使用して、各宛先テーブルの列をソースメッセージのフィールドにマッピングします。たとえば、メッセージから userid フィールドをテーブルの列にマッピングするには、式 $.data.value.userid を使用します。

          データベースユーザー名

          データベースアカウントのユーザー名を入力します。

          user

          データベースパスワード

          データベースアカウントのパスワードを入力します。

          ******

          ネットワーク設定

          • [VPC]: VPC 経由で AnalyticDB に接続します。

          • [Public Network]：パブリックインターネット経由で AnalyticDB に接続します。

          VPC

          VPC

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

          vpc-bp17fapfdj0dwzjkd****

          vSwitch

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

          重要

          vSwitch を選択した後、その vSwitch の CIDR ブロックを AnalyticDB for MySQL インスタンスの IP アドレスホワイトリストに追加する必要があります。詳細については、「IP アドレスホワイトリストを設定する」をご参照ください。

          vsw-bp1gbjhj53hdjdkg****

          セキュリティグループ

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

          test_group

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

    5. ヒントダイアログボックスでメッセージを読み、OKをクリックします。

      タスクの開始には 30 ～ 60 秒かかる場合があります。タスクリストページのStatus列で進行状況をモニターできます。

    ステップ 3：AnalyticDB シンクコネクタの検証

    1. タスクリスト ページで、タスクを見つけ、イベントソース 列にあるソース トピック名をクリックします。

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

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

      説明

      メッセージ本文は JSON 形式である必要があります。データマッピングルールで指定されたフィールドが抽出され、宛先テーブルの対応する列に書き込まれます。

      [メッセージの送受信を開始] ダイアログボックスで、[コンソール] タブを選択します。[メッセージキー] を adb-sink-k1 に、[メッセージ本文] を {"userid":"xiaoming","source":"shanghai"} に設定します。[特定のパーティションに送信] については、[いいえ] を選択し、[OK] をクリックします。

    4. タスクリスト ページで、目的のタスクを見つけ、イベントターゲット 列の宛先インスタンス名をクリックします。

    5. インスタンスのBasic Information ページで、右上隅にあるログインデータベース をクリックします。

    6. データ管理 (DMS) コンソールで、次の SQL ステートメントを実行してテーブル内のすべてのデータをクエリします。

      SELECT * FROM  adb_sink_table;

      クエリを実行すると、userid が xiaoming で source が shanghai のレコードが返されます。これにより、データが宛先テーブルに正常に書き込まれたことが確認されます。