すべてのプロダクト
Search

このプロダクト

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

    ドキュメントセンター

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

    最終更新日:Jun 23, 2026

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

    前提条件

    詳細な手順については、「sink コネクタの前提条件」をご参照ください。

    注意事項

    MaxCompute のパーティション機能を使用するには、テーブルを作成する際に、データ型が `STRING` の `time` という名前のパーティション列を追加で作成する必要があります。

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

    MaxCompute クライアントを使用してテーブルを作成します。詳細については、「テーブルの作成」をご参照ください。

    このチュートリアルでは、`kafka_to_maxcompute` という名前のテーブルを例として使用します。このテーブルには 3 つの列が含まれ、パーティション機能を使用します。次の SQL ステートメントでテーブルを作成します。

    CREATE TABLE IF NOT EXISTS kafka_to_maxcompute(topic STRING,valueName STRING,valueAge BIGINT) PARTITIONED by (time STRING);

    パーティション機能を使用しない場合は、次のステートメントを使用します。

    CREATE TABLE IF NOT EXISTS kafka_to_maxcompute(topic STRING,valueName STRING,valueAge BIGINT);

    ステートメントが正常に実行されると、次の結果が返されます。

    [Tables] ページで、作成された kafka_to_maxcompute テーブルの情報を表示します。パーティションタイプはパーティションテーブル、テーブルタイプは内部テーブルです。テーブルスキーマには、topic (string)、valueName (string)、valueAge (bigint) の 3 つのフィールドが含まれています。いずれもプライマリキーではありません。パーティションフィールドは time (string) です。

    ステップ 2：コネクタの作成と起動

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

    2. 左側のナビゲーションウィンドウで、[コネクタイエコシステム連携] > [タスク] を選択します。

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

    4. タスクの作成 ページで、タスク名 と 説明 パラメーターを設定します。その後、画面の指示に従って他のパラメーターを設定します。

      • タスクの作成

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

          パラメーター

          説明

          例

          Region

          ApsaraMQ for Kafka インスタンスが存在するリージョン。

          中国 (杭州)

          ApsaraMQ For Kafka インスタンス

          ルーティングしたいデータが生成される ApsaraMQ for Kafka インスタンスの ID。

          alikafka_post-cn-9hdsbdhd****

          Topic

          ルーティングしたいデータが生成される ApsaraMQ for Kafka インスタンス上の Topic。

          guide-sink-topic

          グループ ID

          ルーティングしたいデータが生成される ApsaraMQ for Kafka インスタンス上のグループの ID。

          • [すばやく作成]：システムは自動的に GID_EVENTBRIDGE_xxx 形式の ID を持つグループを作成します。

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

          既存のグループを使用

          Consumer Offset

          • [最新のオフセット]：メッセージは最新のオフセットから消費されます。

          • [最も早いオフセット]：メッセージは最も早いオフセットから消費されます。

          最新のオフセット

          ネットワーク設定

          クロスボーダーデータ伝送が必要な場合は、[セルフマネージドインターネット] を選択します。その他の場合は、[基本ネットワーク] を選択します。

          ベーシックネットワーク

          [データ形式]

          データ形式機能は、ソースから配信されたバイナリデータを特定のデータ形式にエンコードするために使用されます。複数のデータ形式がサポートされています。エンコーディングに特別な要件がない場合は、値として Json を指定します。

          • Json：バイナリデータを UTF-8 に基づいて JSON 形式にエンコードし、ペイロードに配置します。

          • Text：デフォルトの形式。バイナリデータを UTF-8 に基づいて文字列にエンコードし、ペイロードに配置します。

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

          テキスト

          一括プッシュの件数

          [詳細設定] パラメーターです。バッチで送信するメッセージの最大数。蓄積されたメッセージ数がこの値に達した場合にのみリクエストが送信されます。値は 1 から 10,000 までの整数である必要があります。

          2000

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

          [詳細設定] パラメーターです。関数が呼び出される間隔。システムはこの間隔でメッセージを集約し、Function Compute に送信します。値は 0 から 15 までの整数である必要があります。単位は秒です。値が 0 の場合は、メッセージがすぐに配信されることを意味します。

          3

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

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

        4. Sink (ターゲット) ステップで、サービスタイプ を [acs.maxcompute] に設定し、次のパラメーターを設定します。

          パラメーター

          説明

          例

          AccessKey ID

          ご利用の Alibaba Cloud アカウントの AccessKey ID。MaxCompute サービスへのアクセスに使用されます。

          yourAccessKeyID

          AccessKey Secret

          ご利用の Alibaba Cloud アカウントの AccessKey Secret。

          yourAccessKeySecret

          MaxCompute プロジェクト名

          既存の MaxCompute プロジェクトを選択します。

          test_compute

          MaxCompute テーブル名

          既存の MaxCompute テーブルを選択します。

          kafka_to_maxcompute

          MaxCompute テーブルの入力パラメータ

          テーブルを選択すると、その列名とデータ型が表示されます。各列に対して [値抽出ルール] を設定するだけです。次のコードはメッセージの例を示しています。この例では、`topic` 列の値は `topic` フィールドから抽出されます。したがって、[値抽出ルール] は $.topic として定義されます。

          {
  'data': {
    'topic': 't_test',
    'partition': 2,
    'offset': 1,
    'timestamp': 1717048990499,
    'headers': {
      'headers': [],
      'isReadOnly': False
    },
    'key': 'MaxCompute-K1',
    'value': 'MaxCompute-V1'
  },
  'id': '9b05fc19-9838-4990-bb49-ddb942307d3f-2-1',
  'source': 'acs:alikafka',
  'specversion': '1.0',
  'type': 'alikafka:Topic:Message',
  'datacontenttype': 'application/json; charset=utf-8',
  'time': '2024-05-30T06:03:10.499Z',
  'aliyunaccountid': '1413397765616316'
}

          topic： $.data.topic

          valuename： $.data.value

          valueage： $.data.offset

          パーティション設定

          • 無効にする：パーティション機能を無効にします。

          • 有効：パーティション機能を有効にします。

            パーティションを有効にする場合は、パーティション値などのパラメーターを設定する必要があります。

            • パーティション値は、時間変数 {yyyy}、{MM}、{dd}、{HH}、および {mm} をサポートします。これらはそれぞれ年、月、日、時、分を表します。時間変数は大文字と小文字を区別します。

            • パーティション値は定数にすることもできます。

          有効

          {yyyy}-{MM}-{dd}.{HH}:{mm}.suffix

          [ネットワーク構成]

          • VPC：VPC を介して Kafka メッセージを MaxCompute に配信します。

          • Public Network：インターネットを介して Kafka メッセージを MaxCompute に配信します。

          インターネット

          VPC

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

          vpc-bp17fapfdj0dwzjkd****

          vSwitch

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

          vsw-bp1gbjhj53hdjdkg****

          セキュリティグループ

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

          test_group

      • タスクプロパティ

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

    5. 上記の設定が完了したら、保存 をクリックします。タスクリスト ページで、作成した MaxCompute sink コネクタタスクを見つけます。ステータス 列には 起動中 と表示されます。ステータスが 実行中 に変わると、作成プロセスは完了です。

    ステップ 3：コネクタのテスト

    1. タスクリスト ページで、MaxCompute sink コネクタを見つけ、イベントソース 列のソース Topic をクリックします。

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

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

      [コンソール] タブで、[メッセージキー] を MaxCompute-K1 に、[メッセージ本文] を MaxCompute-V1 に設定し、[指定したパーティションに送信] を [いいえ] に設定します。

    4. MaxCompute コンソールに移動し、次の SQL ステートメントを実行してパーティション情報を表示します。

      show PARTITIONS kafka_to_maxcompute;

      次の結果が返されます。

      OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
time=2024-05-31.16:37.suffix
OK
2024-05-31 16:42:49 INFO ==================================================================
2024-05-31 16:42:49 INFO Exit code of the Shell command 0
2024-05-31 16:42:49 INFO --- Invocation of Shell command completed ---
2024-05-31 16:42:49 INFO Shell run successfully!
2024-05-31 16:42:49 INFO Current task status: FINISH
2024-05-31 16:42:49 INFO Cost time is: 1.411s

    5. パーティション情報に基づいて、次のステートメントを実行してパーティション内のデータを表示します。

      SELECT * FROM kafka_to_maxcompute WHERE time="2024-05-31.16:37.suffix";

      クエリは 1 つのデータレコードを返します。列の値は次のとおりです：`topic` は xxx (マスク済み)、`valueName` は MaxCompute-V1、`valueAge` は 4、`time` は 2024-05-31.16:37.suffix です。これは、データが ApsaraMQ for Kafka から MaxCompute のパーティションテーブルに正常に書き込まれたことを示します。