すべてのプロダクト
Search

このプロダクト

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

    ドキュメントセンター

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

    最終更新日:Jun 22, 2026

    このトピックでは、OSS シンクコネクターを作成する方法を説明します。このコネクターを使用して、ApsaraMQ for Kafka のソース トピックから OSS オブジェクトにデータをエクスポートできます。

    前提条件

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

    注意事項

    1. コネクターは、イベントの作成時間ではなく、イベントの処理時間に基づいてデータをパーティション分割します。時間ベースのパーティション分割を使用する場合、時間境界に近いデータが次の時間枠のディレクトリに配信されることがあります。

    2. ダーティデータの処理：カスタムパーティションパスまたはファイルコンテンツに JSONPath 式を設定し、受信データがルールに一致しない場合、コネクターはこのダーティデータをバッチ処理ポリシーに従ってバケット内の invalidRuleData/ ディレクトリに配信します。バケット内にこのディレクトリが見つかった場合は、JSONPath 式を検証し、コンシューマーがデータを見逃していないことを確認してください。

    3. エンドツーエンドのレイテンシーは、数秒から数分かかる場合があります。

    4. カスタムパーティションパスまたはファイルコンテンツに設定された JSONPath ルールで Kafka ソースのメッセージ本文からデータを抽出する場合、ソース側でコンテンツを JSON 形式にエンコードする必要があります。

    5. コネクターは、オブジェクトへの追記によって、アップストリームソースから OSS にデータをリアルタイムで書き込みます。そのため、単一のパーティションパスでは、最新の可視オブジェクトは通常まだ書き込み中であり、最終状態ではありません。このデータを消費する際は注意してください。

    課金

    コネクタータスクは Function Compute 上で実行されます。データ処理と転送で消費されるコンピューティングリソースは、Function Compute の料金に基づいて課金されます。詳細については、「課金の概要」をご参照ください。

    ステップ 1：ターゲットサービスリソースの作成

    OSS コンソールでバケットを作成します。詳細については、「コンソールでのバケットの作成」をご参照ください。

    この例では、oss-sink-connector-bucket という名前のバケットを使用します。

    ステップ 2：OSS シンクコネクターの作成と開始

    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 (ターゲット) ステップで、サービスタイプ で Object Storage Service (OSS) を選択し、次のパラメーターを設定してから 保存 をクリックします。

          パラメーター

          説明

          例

          OSS バケット

          作成した OSS バケット。

          重要

          • 指定したバケットが存在し、タスクの実行中に削除されないことを確認してください。

          • バケットのストレージクラスは、標準または低頻度アクセス (IA) である必要があります。アーカイブストレージはサポートされていません。

          • OSS シンクコネクタータスクを作成すると、システムはバケットのルートディレクトリに .tmp/ システムファイルパスを生成します。このパスを削除したり、その中の OSS オブジェクトを使用したりしないでください。

          oss-sink-connector-bucket

          [保存パス]

          OSS オブジェクトキーは、パスと名前で構成されます。たとえば、オブジェクトキーが a/b/c/a.txt の場合、パスは a/b/c/、名前は a.txt です。パーティションパスはカスタマイズできます。名前はコネクターによって {ミリ秒単位の Unix タイムスタンプ}_{8 文字のランダムな文字列} の形式で自動的に生成されます (例：1655952591589_abcdefgh)。

          • この項目を空白にするか、/ に設定すると、パーティション分割は適用されず、データはバケットのルートディレクトリに保存されます。

          • 時間変数をサポートします：{yyyy}、{MM}、{dd}、{HH} はそれぞれ年、月、日、時を表します。これらの変数は大文字と小文字を区別します。

          • JSONPath 式を使用してパスをカスタマイズすることもできます。例：{$.data.topic} や {$.data.partition}。JSONPath 変数は有効な JSONPath 式である必要があります。OSS パスの制約により、JSONPath を使用して抽出された値は int 型または string 型である必要があります。値には標準の UTF-8 文字のみを含める必要があり、スペース、..、絵文字、/、または \ を含めることはできません。そうしないと、データの書き込み例外が発生する可能性があります。

          • 定数をサポートします。

            説明

            パーティション分割は、データを論理的にグループ化し、単一パス内の過剰な数の小さなオブジェクトによるパフォーマンスの問題を防ぐのに役立ちます。

            コネクターのスループットはパーティション数に正の相関があります。パーティション分割が少ないか、まったくない場合、スループットが低くなり、ソースでデータのバックログが発生する可能性があります。パーティションが多すぎると、データの断片化、書き込み操作の増加、および過剰な数の小さなオブジェクトにつながる可能性があります。したがって、適切なパーティション分割戦略が重要です。次の推奨事項を考慮してください：

            • Kafka ソース：時間とパーティションの両方によるパーティション分割が可能です。パフォーマンスが不十分な場合は、Kafka のパーティション数を増やすことで、コネクターのスループットを間接的に向上させることができます。例：prefix/{yyyy}/{MM}/{dd}/{HH}/{$.data.partition}/

            • ビジネス グループ化：特定のビジネスフィールドでデータをパーティション分割します。スループット率は、このフィールドの一意の値の数によって決まります。例：prefixV2/{$.data.body.field}/

            複数のタスクが同じディレクトリに書き込むことでデータが混在するのを避けるために、タスクごとに異なる定数プレフィックスを使用することを推奨します。

          alikafka_{$.data.topic}/{$.data.partition}/{yyyy}/{MM}/{dd}/{HH}

          [バッチ集計オブジェクトサイズ]

          オブジェクト集計のターゲットサイズ。値は MB 単位です。有効な値の範囲：1～1,024。

          説明

          • コネクターは最大 16 MB のバッチでデータを書き込みます。その結果、最終的なオブジェクトサイズは設定値より最大 16 MB 大きくなる可能性があります。

          • 高トラフィックのシナリオでは、バッチ集計オブジェクトサイズを数百メガバイト (例：128 MB または 512 MB) に、時間枠を時間単位 (例：60 分または 120 分) に設定することを推奨します。

          5

          [バッチ集計時間枠]

          集計の時間枠。値は分単位です。有効な値の範囲：1～1,440。

          1

          [ファイルの圧縮]

          • [No Compression Required]：拡張子なしの OSS オブジェクトを生成します。

          • [GZIP]：.gz 拡張子が付いたオブジェクトを生成します。

          • [Snappy]：.snappy 拡張子が付いたオブジェクトを生成します。

          • [Zstd]：.zstd 拡張子が付いたオブジェクトを生成します。

          圧縮オプションを選択した場合、コネクターは圧縮前のサイズに基づいてデータをバッチ処理します。その結果、OSS のオブジェクトサイズは設定された [Batch aggregation object size] よりも小さくなります。展開後、サイズは設定値に近くなります。

          [No Compression Required]

          [ファイルの内容]

          • [Complete Data]：コネクターは元のメッセージを CloudEvents プロトコルでラップします。完全なデータには、CloudEvents プロトコルラッパーを含むデータが含まれます。次の例では、data フィールドにメッセージデータが含まれ、他のフィールドは CloudEvents プロトコルによって追加されたメタデータです。

            {
  "specversion": "1.0",
  "id": "8e215af8-ca18-4249-8645-f96c1026****",
  "source": "acs:alikafka",
  "type": "alikafka:Topic:Message",
  "subject": "acs:alikafka:alikafka_pre-cn-i7m2msb9****:topic:****",
  "datacontenttype": "application/json; charset=utf-8",
  "time": "2022-06-23T02:49:51.589Z",
  "aliyunaccountid": "182572506381****",
  "data": {
    "topic": "****",
    "partition": 7,
    "offset": 25,
    "timestamp": 1655952591589,
    "headers": {
      "headers": [],
      "isReadOnly": false
    },
    "key": "keytest",
    "value": "hello kafka msg"
  }
}

          • [Data Extraction]：JSONPath 式を使用して抽出されたデータの一部のみを配信します。たとえば、$.data を指定すると、data フィールドの値のみが OSS に配信されます。

          ストレージコストを節約し、効率を向上させるには、$.data 式を指定して [Data Extraction] を使用します。これにより、CloudEvents ラッパーを省略して、元のソースメッセージのみが OSS に配信されます。

          [Data Extraction]

          $.data

      • タスクのプロパティ

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

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

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

      タスクの開始には 30～60 秒かかります。タスクリスト ページの Status 列で進捗を監視できます。

    ステップ 3：OSS シンクコネクターのテスト

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

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

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

      [Console] タブを選択します。[Message Key] フィールドに oss-sink-k2 と入力します。[Message Content] フィールドに oss-sink-v2 と入力します。[Send to Specified Partition] で [No] を選択します。

    4. タスクリスト ページで、タスクの イベントターゲット 列にあるターゲットバケットをクリックします。

    5. バケットページで、左側のナビゲーションペインから オブジェクト管理 > [Objects] を選択します。

      • tmp ディレクトリ：コネクターが依存するシステムパスです。このパス内の OSS オブジェクトを削除したり、使用したりしないでください。

      • データファイルディレクトリ：サブディレクトリはタスクのパーティションパスルールに基づいて生成され、データオブジェクトは最も深いディレクトリにアップロードされます。

      この例では、パンくずリストのパスは /alikafka_<topic>/<partition>/2023/04/18/02/ です。これは、サブディレクトリがトピック名、パーティション番号、および年/月/日/時の構造に基づいて作成されることを示しています。最も深いディレクトリには、.oss_meta_file や partition_3_of... などのデータオブジェクトが含まれます。

    6. オブジェクトの右側にある 操作 列で、图标 > ダウンロード を選択します。

    7. ダウンロードしたファイルを開き、メッセージの内容を確認します。

      {"topic":"guide-oss-sink-topic","partition":0,"offset":0,"timestamp":1681378474218,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}
{"topic":"guide-oss-sink-topic","partition":0,"offset":1,"timestamp":1681378491498,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}
{"topic":"guide-oss-sink-topic","partition":0,"offset":2,"timestamp":1681378492515,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}

      出力には複数のメッセージが含まれ、各メッセージは新しい行に JSON オブジェクトとしてフォーマットされています。