すべてのプロダクト
Search

このプロダクト

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

    ドキュメントセンター

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

    最終更新日:Jun 22, 2026

    Function Compute (FC) sink コネクタを作成して、ご利用の ApsaraMQ for Kafka インスタンスのソース Topic から FC の関数にデータをエクスポートします。

    前提条件

    FC sink コネクタを作成する前に、次の前提条件が満たされていることを確認してください。

    • ApsaraMQ for Kafka

      • ApsaraMQ for Kafka インスタンスのコネクタ機能を有効にします。詳細については、「コネクタの有効化」をご参照ください。

      • ApsaraMQ for Kafka インスタンスのソース Topic を作成します。詳細については、「ステップ 1：Topic の作成」をご参照ください。

        この例では、fc-test-input という名前の Topic を使用します。

    • Function Compute

      • FC で関数を作成します。詳細については、「関数の作成」をご参照ください。

        重要

        関数はイベント関数である必要があります。

        このガイドでは、hello_world という名前のイベント関数を例として使用します。この関数は guide-hello_world サービス内にあり、Python 実行環境で実行されます。関数のサンプルコードは次のとおりです。

        # -*- coding: utf-8 -*-
import logging
# 初期化機能を有効にするには
# 初期化関数を次のように実装します：
# def initializer(context):
#   logger = logging.getLogger()
#   logger.info('initializing')
def handler(event, context):
  logger = logging.getLogger()
  logger.info('hello world:' + bytes.decode(event))
  return 'hello world:' + bytes.decode(event)

    • オプション： EventBridge

      説明

      この手順は、ご利用の ApsaraMQ for Kafka インスタンスが中国 (杭州) または中国 (成都) リージョンにある場合にのみ必要です。

    注意事項

    • ApsaraMQ for Kafka インスタンスのソース Topic から FC の関数にデータをエクスポートできるのは、両者が同じリージョンにある場合のみです。コネクタの制限事項の詳細については、「制限事項」をご参照ください。

    • ご利用の ApsaraMQ for Kafka インスタンスが中国 (杭州) または中国 (成都) リージョンにある場合、コネクタは EventBridge にデプロイされます。

      • EventBridge は現在無料です。詳細については、「課金」をご参照ください。

      • コネクタを作成すると、EventBridge は次のサービスリンクロールを自動的に作成します：AliyunServiceRoleForEventBridgeSourceKafka および AliyunServiceRoleForEventBridgeConnectVPC。

        • サービスリンクロールが作成されていない場合、EventBridge は対応するサービスリンクロールを自動的に作成し、EventBridge がこのロールを使用して ApsaraMQ for Kafka および VPC にアクセスできるようにします。

        • これらのサービスリンクロールが既に存在する場合、EventBridge はそれらを再作成しません。

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

      • 現在、EventBridge にデプロイされたタスクのランタイムログは表示できません。コネクタタスクが完了したら、ソース Topic のコンシューマーグループの消費ステータスを表示して進行状況を確認します。詳細については、「コンシューマーのステータスを表示する」をご参照ください。

    手順

    ApsaraMQ for Kafka インスタンスのソース Topic から FC の関数にデータをエクスポートするには、FC sink コネクタを使用します。

    1. オプション： FC sink コネクタのクロスリージョンアクセスを有効にする

      重要

      クロスリージョンアクセスが不要な場合は、この手順をスキップしてください。

      FC sink コネクタのインターネットアクセスを有効にする

    2. オプション： FC sink コネクタのクロスアカウントアクセスを有効にする

      重要

      クロスアカウントアクセスが不要な場合は、この手順をスキップしてください。

    3. オプション： FC sink コネクタで必要な Topic とコンシューマーグループを作成する

      重要

      • Topic とコンシューマーグループの名前をカスタマイズする必要がない場合は、この手順をスキップできます。

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

      1. FC sink コネクタで必要な Topic の作成

      2. FC sink コネクタで必要なコンシューマーグループの作成

    4. FC sink コネクタの作成とデプロイ

    5. 結果の確認

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

      2. 関数ログの表示

    FC sink コネクタのインターネットアクセスを有効にする

    FC sink コネクタが他のリージョンの Alibaba Cloud サービスにアクセスする必要がある場合は、インターネットアクセスを有効にする必要があります。詳細については、「コネクタのインターネットアクセスを有効にする」をご参照ください。

    カスタムポリシーの作成

    ターゲットアカウントで、FC へのアクセスを許可するカスタムポリシーを作成します。

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

    2. 左側のナビゲーションウィンドウで、[権限] > [ポリシー] を選択します。

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

    4. [ポリシーの作成] ページで、ポリシーを設定します。

      1. [JSON] タブで、ポリシースクリプトを入力し、[次へ] をクリックします。

        次のポリシースクリプトは、FC にアクセスする権限を付与します。

        {
    "Version": "1",
    "Statement": [
        {
            "Action": [
                "fc:InvokeFunction",
                "fc:GetFunction"
            ],
            "Resource": "*",
            "Effect": "Allow"
        }
    ]
}

      2. 基本情報で、名前に KafkaConnectorFcAccess を入力します。

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

    RAM ロールの作成

    ターゲットアカウントで RAM ロールを作成します。RAM ロールの作成時に、信頼できるサービスとして ApsaraMQ for Kafka を選択することはできません。そのため、サポートされている別のサービスを選択し、ロールの作成後に信頼ポリシーを手動で変更する必要があります。

    1. 左側のナビゲーションウィンドウで、[ID] > [ロール] を選択します。

    2. [ロール] ページで、[ロールの作成] をクリックします。

    3. [ロールの作成] パネルで、ロールを設定します。

      1. 信頼できるエンティティタイプとして [Alibaba Cloud サービス] を選択し、[次へ] をクリックします。

      2. [ロールタイプ] セクションで、[通常のサービスロール] を選択します。[ロール名] に AliyunKafkaConnectorRole と入力します。[信頼されたサービスの選択] ドロップダウンリストから [Function Compute] を選択し、[完了] をクリックします。

    4. [ロール] ページで、[AliyunKafkaConnectorRole] を見つけてクリックします。

    5. [AliyunKafkaConnectorRole] 詳細ページで、[信頼ポリシー管理] タブをクリックし、[信頼ポリシーの編集] をクリックします。

    6. [信頼ポリシーの編集] パネルで、スクリプト内の fc を alikafka に置き換えて、[OK] をクリックします。

      変更を保存した後、[AliyunKafkaConnectorRole] の [信頼ポリシー管理] タブで、信頼ポリシーの Service パラメーターが alikafka.aliyuncs.com に更新され、Action パラメーターが sts:AssumeRole に設定されていることを確認します。

    権限の追加

    ターゲットアカウントで、RAM ロールに FC へのアクセス権限を付与します。

    1. 左側のナビゲーションウィンドウで、[ID] > [ロール] に移動します。

    2. [ロール] ページで、[AliyunKafkaConnectorRole] を見つけ、[操作] 列の [権限の追加] をクリックします。

    3. [権限の追加] パネルで、[KafkaConnectorFcAccess] ポリシーを追加します。

      1. [ポリシーの選択] セクションで、[カスタムポリシー] を選択します。

      2. [認証ポリシー名] リストで、[KafkaConnectorFcAccess] を見つけてクリックします。

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

      4. [完了] をクリックします。

    FC sink コネクタ用の Topic の作成

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

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

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

      重要

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

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

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

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

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

      パラメーター

      説明

      例

      名前

      Topic の名前。

      説明

      Kafka は、アンダースコア (xxx_xxx) とピリオド (xxx.xxx) を含む Topic 名を同一と見なします。重複した Topic を作成しようとすると、システムはエラーを報告します。

      demo

      記述

      Topic の簡単な説明。

      demo test

      パーティションの数

      Topic のパーティション数。

      12

      ストレージエンジン

      説明

      現在、ストレージエンジンを選択できるのは、サーバーレスではない Professional Edition インスタンスのみです。他のインスタンスタイプではこのオプションは利用できず、デフォルトで [クラウドストレージ] が使用されます。

      Topic 内のメッセージのストレージエンジン。

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

      • クラウドストレージ: このエンジンは、基盤ストレージとして Alibaba Cloud ディスクを使用し、分散型の 3 レプリカメカニズムにより、高性能、低レイテンシー、高信頼性を提供します。インスタンスの 仕様タイプ が Standard Edition (High Write) の場合、クラウドストレージ のみ使用できます。

      • ローカルストレージ：このエンジンは、ネイティブの Kafka 同期レプリカ (ISR) レプリケーションアルゴリズムと分散 3 レプリカメカニズムを使用します。

      クラウドストレージ

      メッセージタイプ

      Topic 内のメッセージのタイプ。

      • 通常のメッセージ: デフォルトでは、Kafka は同じキーを持つメッセージを同じパーティションに配信し、送信された順序で保存します。クラスター内のノードで障害が発生した場合、メッセージの順序が乱れることがあります。ストレージエンジン を クラウドストレージ に設定すると、システムはデフォルトで 通常のメッセージ を選択します。

      • パーティション順位メッセージ: デフォルトで、Kafka は同じキーを持つメッセージを同じパーティションに配信し、送信された順序で保存します。クラスターノードに障害が発生した場合でも、パーティション内のメッセージの順序は保証されます。ただし、一部のパーティションが回復するまで、それらのパーティションへのメッセージの送信が失敗することがあります。ストレージエンジン を ローカルストレージ に設定すると、システムはデフォルトで パーティション順位メッセージ を選択します。

      通常のメッセージ

      ログリリースポリシー

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

      ストレージエンジンとしてローカルストレージを選択した場合 (現在、Professional Edition インスタンスのみがローカル記憶域をサポートしており、このオプションは Standard Edition インスタンスでは利用できません)、ログリリースポリシーを設定する必要があります。

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

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

      • Compact：Kafka ログ圧縮クリーンアップポリシーを使用します。ログ圧縮により、システムは各メッセージキーの最新の値を保持します。これは主に、システムクラッシュ後の状態復元やシステム再起動後のキャッシュ再読み込みなどのシナリオで使用されます。たとえば、Kafka Connect や Confluent Schema Registry は、圧縮された Topic を使用してシステムの状態と設定データを保存します。

        重要

        圧縮された Topic は、通常、Kafka Connect や Confluent Schema Registry などの特定の生態系コンポーネントにのみ使用されます。一般的なメッセージの生成と消費に使用される Topic にはこのプロパティを設定しないでください。詳細については、「ApsaraMQ for Kafka デモライブラリ」をご参照ください。

      Compact

      タグ

      Topic のタグ。

      demo

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

    FC sink コネクタのコンシューマーグループの作成

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

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

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

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

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

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

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

      作成されると、コンシューマーグループはGroup の管理ページのリストに表示されます。

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

    FC sink コネクタを作成してデプロイし、ApsaraMQ for Kafka から FC にデータを同期します。

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

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

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

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

    5. Connector の作成 ウィザードでは、以下の手順を完了します。

      1. 基本情報の設定 タブで、必要に応じて次のパラメーターを設定し、次へ をクリックします。

        パラメーター

        説明

        例

        名前

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

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

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

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

        kafka-fc-sink

        インスタンス

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

        demo alikafka_post-cn-st21p8vj****

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

        説明

        すでに Topic とコンシューマーグループを作成している場合は、手動でのリソース作成を選択し、既存のリソースの情報を入力します。それ以外の場合は、自動でのリソース作成を選択します。

        表 1. ソースサービスを構成するためのパラメーター

        パラメーター

        説明

        例

        データソース Topic

        データが同期されるソース Topic。

        fc-test-input

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

        ソース Topic の同時コンシューマースレッド数。デフォルト値：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***

        失敗の処理

        パーティションからのメッセージの配信が失敗した場合の動作を制御します。このパラメーターは、実行環境の設定 をクリックすると表示されます。有効な値:

        • サブスクリプションの継続：パーティションからの消費を続行し、エラーをログに記録します。

        • サブスクリプションの停止：パーティションからの消費を停止し、エラーをログに記録します。

        説明

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

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

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

        リソースの作成方法

        コネクタで必要となる Topic と コンシューマーグループ を作成するためのメソッドです。このパラメーターは、実行環境の設定 をクリックすると表示されます。

        • 自動作成

        • 手動で作成します

        自動作成

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

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

        connect-kafka-fc-sink

        タスクサイトの Topic

        コンシューマオフセットを格納するために使用される Topic です。このパラメーターは、実行環境の設定 をクリックすると表示されます。

        • Topic 名は connect-offset で始まる必要があります。

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

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

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

        connect-offset-kafka-fc-sink

        タスク設定の Topic

        タスク構成を格納するために使用される Topic です。このパラメーターは、実行環境の設定 をクリックすると表示されます。

        • Topic：Topic 名は connect-config で始まることを推奨します。

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

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

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

        connect-config-kafka-fc-sink

        タスクステータスの Topic

        タスクステータスを格納するために使用される Topic です。このパラメーターは実行環境の設定をクリックすると表示されます。

        • Topic：Topic 名は connect-status で始まることを推奨します。

        • パーティション：パーティション数を 6 に設定することを推奨します。

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

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

        connect-status-kafka-fc-sink

        デッドレターキューの Topic

        Connect フレームワークからのエラーデータを格納するために使用される Topic です。このパラメーターは、実行環境の設定 をクリックすると表示されます。Topic リソースを節約するために、このパラメーターと エラーデータ Topic パラメーターの両方に同じ Topic を使用できます。

        • Topic：Topic 名は connect-error で始まることを推奨します。

        • パーティション：パーティション数を 6 に設定することを推奨します。

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

        connect-error-kafka-fc-sink

        例外データ Topic

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

        • Topic：Topic 名は connect-error で始まることを推奨します。

        • パーティション：パーティション数を 6 に設定することを推奨します。

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

        connect-error-kafka-fc-sink

      3. ターゲットサービスの設定 タブで、[ターゲットサービス]に[Function Compute]を選択し、次のパラメーターを設定して、作成をクリックします。

        説明

        コネクタのインスタンスが中国 (杭州) または中国 (成都) リージョンにあり、[ターゲットサービス] として Function Compute を選択した場合、AliyunServiceRoleForEventBridgeSourceKafka および AliyunServiceRoleForEventBridgeConnectVPC サービスリンクロールを作成するための [サービス認証] ダイアログボックスが表示されます。 [サービス認証] ダイアログボックスで、[確認] をクリックし、次のパラメーターを設定してから 作成 をクリックします。 サービスリンクロールがすでに作成されている場合、[サービス認証] ダイアログボックスは表示されません。

        パラメーター

        説明

        例

        アカウントまたはリージョン間をクロスオーバーするかどうか

        FC Sink Connector が、異なるアカウントまたはリージョンにある Function Compute サービスにデータを同期するかどうかを指定します。デフォルト値: いいえ。有効な値:

        • いいえ：同一アカウント、同一リージョンモード。

        • はい：クロスアカウント、クロスリージョン、またはその両方。

        いいえ

        サービスリージョン

        FC サービスのリージョン。デフォルトでは、これは FC sink コネクタのリージョンです。リージョンをまたいでデータを同期するには、コネクタのパブリックアクセスを有効にしてから、宛先リージョンを選択する必要があります。詳細については、「FC sink コネクタのパブリックアクセスを有効にする」をご参照ください。

        重要

        アカウントまたはリージョン間をクロスオーバーするかどうか を はい に設定すると、サービスリージョン が表示されます。

        cn-hangzhou

        サービスのアクセスポイント

        FC サービスのエンドポイント。FC コンソールの [概要] ページの [共通情報] セクションからエンドポイントを取得できます。

        • 内部エンドポイント：低遅延のため推奨されます。ご利用の ApsaraMQ for Kafka インスタンスと FC サービスが同じリージョンにある場合は、このタイプのエンドポイントを使用します。

        • パブリックエンドポイント：遅延が大きいため推奨されません。ご利用の ApsaraMQ for Kafka インスタンスと FC サービスが異なるリージョンにある場合は、このタイプのエンドポイントを使用します。パブリックエンドポイントを使用するには、コネクタのパブリックアクセスを有効にする必要があります。詳細については、「FC sink コネクタのパブリックアクセスを有効にする」をご参照ください。

        重要

        アカウントまたはリージョン間をクロスオーバーするかどうか が はい に設定されている場合、サービスのアクセスポイント が表示されます。

        http://188***.cn-hangzhou.fc.aliyuncs.com

        サービスアカウント

        FC サービスが属する Alibaba Cloud アカウントの ID。ID は、FC コンソールの [概要] ページの [共通情報] セクションから取得できます。

        重要

        アカウントまたはリージョン間をクロスオーバーするかどうか が はい に設定されている場合、サービスアカウント が表示されます。

        188***

        権限が付与されたロール名

        ApsaraMQ for Kafka が FC サービスにアクセスするために引き受けるRAM ロールの名前。

        重要

        アカウントまたはリージョン間をクロスオーバーするかどうか が はい に設定されている場合、権限が付与されたロール名 が表示されます。

        AliyunKafkaConnectorRole

        サービス名

        FC のサービスの名前。

        guide-hello_world

        関数名

        FC サービス内の関数の名前。

        hello_world

        バージョンまたはエイリアス

        FC サービスのバージョンまたはエイリアス。

        重要

        • アカウントまたはリージョン間をクロスオーバーするかどうか が いいえ に設定されている場合、指定されたバージョン または エイリアスの指定 のいずれかを選択する必要があります。

        • アカウントまたはリージョン間をクロスオーバーするかどうか が はい に設定されている場合、サービスバージョンまたはエイリアスを手動で入力する必要があります。

        LATEST

        サービスバージョン

        FC サービスのバージョン。

        重要

        アカウントまたはリージョン間をクロスオーバーするかどうか が いいえ に設定されており、バージョンまたはエイリアス が 指定されたバージョン に設定されている場合、サービスバージョン パラメーターが表示されます。

        LATEST

        サービスのエイリアス

        FC サービスのエイリアス。

        重要

        アカウントまたはリージョン間をクロスオーバーするかどうか が いいえ に設定され、バージョンまたはエイリアス が エイリアスの指定 に設定されているとき、サービスのエイリアス パラメーターが表示されます。

        jy

        送信モード

        メッセージを送信するモード。有効な値：

        • 非同期：推奨。

        • 同期：非推奨。このモードでは、FC によるメッセージ処理が遅いと、ApsaraMQ for Kafka も遅くなります。メッセージのバッチ処理に 5 分以上かかると、ApsaraMQ for Kafka でクライアントのリバランスがトリガーされます。

        非同期

        ロットサイズを送信

        1 つのバッチに含めるメッセージの最大数。コネクタは、この数と基盤となるリクエストサイズの制限 (同期の場合は 6 MB、非同期の場合は 128 KB) の両方を尊重してメッセージをバッチに集約します。たとえば、配信モードが非同期で、バッチサイズが 20 で、18 件のメッセージを送信したい場合、17 件のメッセージの合計サイズが 127 KB で、1 件のメッセージのサイズが 200 KB であるとします。コネクタは 17 件のメッセージを 1 つのバッチに集約して送信します。コネクタは、サイズが 128 KB を超える残りのメッセージを別のバッチで送信します。

        説明

        メッセージ送信時に key を null に設定すると、リクエストに key は含まれません。value を null に設定すると、リクエストに value は含まれません。

        • バッチ内のメッセージの合計サイズがリクエストサイズの制限を超えない場合、リクエストにはメッセージの内容が含まれます。次のコードは、サンプルリクエストを示しています。

          [
    {
        "key":"this is the message's key2",
        "offset":8,
        "overflowFlag":false,
        "partition":4,
        "timestamp":1603785325438,
        "topic":"Test",
        "value":"this is the message's value2",
        "valueSize":28
    },
    {
        "key":"this is the message's key9",
        "offset":9,
        "overflowFlag":false,
        "partition":4,
        "timestamp":1603785325440,
        "topic":"Test",
        "value":"this is the message's value9",
        "valueSize":28
    },
    {
        "key":"this is the message's key12",
        "offset":10,
        "overflowFlag":false,
        "partition":4,
        "timestamp":1603785325442,
        "topic":"Test",
        "value":"this is the message's value12",
        "valueSize":29
    },
    {
        "key":"this is the message's key38",
        "offset":11,
        "overflowFlag":false,
        "partition":4,
        "timestamp":1603785325464,
        "topic":"Test",
        "value":"this is the message's value38",
        "valueSize":29
    }
]

        • 1 つのメッセージのサイズがリクエストサイズの制限を超える場合、リクエストにはメッセージの内容が含まれません。次のコードは、サンプルリクエストを示しています。

          [
    {
        "key":"123",
        "offset":4,
        "overflowFlag":true,
        "partition":0,
        "timestamp":1603779578478,
        "topic":"Test",
        "value":"1",
        "valueSize":272687
    }
]
          説明

          メッセージの内容を取得するには、オフセットに基づいてメッセージをプルする必要があります。

        50

        再試行回数

        メッセージの送信に失敗した後のリトライ回数。デフォルト値は 2 です。値の範囲は 1 から 3 です。メッセージ送信の失敗を引き起こす一部のエラーは、リトライをサポートしていません。エラーコードとリトライサポートのマッピングは次のとおりです。

        • 4XX：429 を除く 4xx エラーではリトライはサポートされていません。

        • 5XX：リトライはサポートされています。

        説明

        • コネクタは InvokeFunction 操作を呼び出して FC にメッセージを送信します。

        • 最大リトライ回数を超えてもメッセージの送信に失敗した場合、メッセージはデッドレターキュー Topic に送信されます。デッドレターキュー Topic 内のメッセージは、FC コネクタタスクを再度トリガーしません。デッドレターキュー Topic のアラートを設定して、そのステータスをリアルタイムで監視し、例外をタイムリーに処理することをお勧めします。

        2

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

    6. Connector タスクリスト ページで、新しく作成したコネクタを見つけ、操作 列の デプロイ をクリックします。

      Function Compute リソースを設定するには、操作 列で 詳細 > 関数の設定 を選択します。Function Compute コンソールにリダイレクトされ、そこで構成を完了します。

    テストメッセージの送信

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

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

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

      • 送信方法 には コンソール を選択します。

        1. メッセージキー フィールドに、demo などのメッセージキーを入力します。

        2. メッセージの内容 フィールドに、{"key": "test"} などのメッセージ本文を入力します。

        3. 指定されたパーティションに送信 では、オプションを選択してください:

          • はい をクリックし、パーティション ID フィールドに 0 などのパーティション ID を入力します。 パーティション ID を確認するには、「パーティションのステータスを表示する」をご参照ください。

          • いいえ をクリックすると、パーティションを指定せずにメッセージが送信されます。

      • 送信方法 で Docker を選択し、Docker コンテナーを実行してサンプルメッセージを生成する セクションのコマンドを実行します。

      • 送信方法 には SDK を選択します。次に、メッセージを送信するため、希望する言語またはフレームワークの SDK と統合方法を選択します。

    関数ログ

    ご利用の ApsaraMQ for Kafka インスタンスのソース Topic にメッセージを送信した後、関数ログをチェックして、関数がそれを受信したことを確認します。詳細については、「ログの設定」をご参照ください。

    テストメッセージがログに表示されます。

    関数詳細ページで、[ログクエリ] タブをクリックし、[高度なクエリ] を選択します。Logstore として function-log を選択します。ログの message フィールドには、Kafka Topic からのメッセージが含まれます。例：[INFO] hello world:[{"key":"1","offset":1,"partition":0,"timestamp":1605598174308,"topic":"fc-test-input","value":"1"}]。これにより、関数が Kafka トリガーからメッセージを正常に受信したことが確認されます。