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

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

• プロジェクトを作成します。詳細については、「プロジェクトの作成」をご参照ください。

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

このトピックでは、guide-sls-sink-project という名前のプロジェクトと guide-sls-sink-logstore という名前の Logstore を例として使用します。

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

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 (ターゲット)] 設定ウィザードで、[サービスタイプ] を [Log Service] に設定し、次のパラメーターを設定して、[保存] をクリックします。

   表 1. シンク (Simple Log Service)

   パラメーター	説明	例
   [ログプロジェクト]	作成した Simple Log Service プロジェクト。	guide-sls-sink-project
   [ログストア]	作成した Simple Log Service Logstore。	guide-sls-sink-logstore
   [ログトピック]	SLS Topic を生成する方法。 [空] [固定値] [一部のデータ] [テンプレート]	Empty
   [ログ内容]	[デフォルト形式 (JSON)] [カスタム形式 (キー/値)]	Default Format (JSON)
   [ロール名]	EventBridge に SLS への書き込み権限を付与します。ロールが存在しない場合は、画面の指示に従って作成してください。	sls_eb

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

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

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

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

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

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

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

   [Console] タブを選択します。[Message Key] を sls-sink-k1 に、[Message Content] を sls-sink-v1 に設定します。[Send to a specified partition] は、[No] を選択します。

4. [タスクリスト] ページで、SLS シンクコネクタータスクを見つけ、[イベントターゲット] 列の宛先プロジェクトをクリックします。

5. Logstore ページで、ログの内容を表示します。

   ログレコードの [content] フィールドは JSON 形式で、key:"sls-sink-k1" と value:"sls-sink-v1" が含まれています。これにより、SLS シンクコネクターが宛先の Logstore にデータを正常に書き込んだことが確認されます。