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

Function Compute

Function Compute (FC) は、フルマネージドでイベント駆動型のサーバーレスコンピューティングサービスです。ユーザーはサーバーを管理することなくコードを記述・アップロードできます。Function Compute はコンピューティングリソースを準備し、コードをエラスティックかつ信頼性高く実行します。詳細については、「Function Compute とは」をご参照ください。

ユースケース

• 関数を使用してビジネスメッセージを処理します。Function Compute 上でメッセージ処理、注文処理、タスク実行などのビジネスロジックを開発・実行します。

• ETL クレンジングなど、メッセージを迅速に処理します。

• 指定された VPC 内で柔軟なスケーリングにより、メッセージをダウンストリームシステムに転送します。

• メッセージングシステムと他の Alibaba Cloud サービスを接続し、メッセージデータを追加のクラウドサービスにルーティングします。

シンクコネクタの作成

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

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

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

4. タスクの作成 パネルで、タスク名 および 説明 を設定し、以下のパラメーターを構成してから、保存 をクリックします。

   • タスク作成

     1. Source (ソース) ステップで、データプロバイダー として Message Queue for Apache Kafka を選択します。以下のパラメーターを構成し、次へ をクリックします。

        パラメーター	説明	例
        リージョン	ソース Message Queue for Apache Kafka インスタンスのリージョンです。	中国 (北京)
        Kafka インスタンス	ソース Message Queue for Apache Kafka インスタンスです。	alikafka_post-cn-jte3****
        Topic	メッセージを消費するトピックです。	demo-topic
        Group ID	ソースインスタンスのコンシューマーグループです。 クイック作成（推奨）：GID_EVENTBRIDGE_xxx 形式の Group ID が自動的に作成されます。 既存のものを使用：既存の Group ID を選択します。他のサービスとこの Group 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 バッチあたりの最大メッセージ数です。蓄積されたメッセージ数がこの値に達するとリクエストが送信されます。有効値：1 ～ 10,000。	100
        バッチプッシュ間隔 (単位：秒)	詳細設定 パラメーターです。メッセージを集約してシンクに送信する間隔（秒単位）です。有効値：0 ～ 15。0 を指定すると、メッセージは即時に配信されます。	3

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

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

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

        パラメーター	説明	例
        関数	既存の Function Compute 関数を選択します。	test
        バージョンとエイリアス	特定のバージョンまたはエイリアスを使用するかどうかを選択します。 指定されたバージョン エイリアスの指定	指定バージョン
        Version	関数のバージョンです。最新バージョンを推奨します。このパラメーターは、バージョンとエイリアス を 指定されたバージョン に設定した場合に必須です。	LATEST
        エイリアス	関数のエイリアスです。このパラメーターは、バージョンとエイリアス を エイリアスの指定 に設定した場合に必須です。	test
        実行方法	呼び出しモード です。有効値：同期 および 非同期。	非同期
        配信形式	Object フォーマット：イベントがオブジェクトとしてダウンストリーム関数に配信されます。 ObjectList フォーマット：イベントがオブジェクトの配列としてダウンストリーム関数に配信されます。	オブジェクト形式
        イベント	EventBridge は JSONPath を使用して、特定のメッセージ本文を抽出し、ターゲットにルーティングします。 完全なデータ データ抽出 固定値 テンプレート	完全データ

   • タスクプロパティ

     タスクの リトライポリシー および デッドレターキュー を構成します。詳細については、「リトライとデッドレターキュー」をご参照ください。

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

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

   タスクを有効化すると、タスクが開始するまでに 30 ～ 60 秒かかる場合があります。タスクリスト ページの Status 列で進捗を監視してください。

その他の操作

タスクリスト ページで対象のタスクを見つけ、操作する 列で以下の操作を実行します。

• タスク詳細の表示： 詳細 をクリックします。タスク ページで、タスクの基本情報、プロパティ、監視メトリクスを確認できます。

• タスク構成の編集： 編集する をクリックします。タスクの編集 パネルで、タスク詳細およびプロパティを変更できます。

• タスクの有効化または一時停止： 有効化する または 無効化 をクリックします。ヒント ダイアログボックスで、OK をクリックします。

• タスクの削除： 削除する をクリックします。ヒント ダイアログボックスで、OK をクリックします。