Simple Log Service:Kafka からのデータインポート

サポートバージョン

Kafka 2.0.0 以降のみがサポートされています。

データインポート設定の作成

1. Simple Log Service コンソールにログインします。

2. データのインポート セクションの データのインポート タブで、[Kafka - データインポート] をクリックします。

3. 送信先プロジェクトと Logstore を選択し、次へ をクリックします。

4. インポート設定を構成します。

   1. [インポート設定]ステップで、以下のパラメーターを設定します。

      パラメーター	説明
      タスク名	データインポート設定の一意な名前。
      表示名	設定の表示名。
      ジョブの説明	インポート設定の説明。
      エンドポイント	Kafka ブートストラップサーバーのエンドポイント。コンマ (,) で区切って複数のエンドポイントを指定できます。 ApsaraMQ for Kafka を使用する場合は、エンドポイントの IP アドレスまたはドメイン名を入力します。 Alibaba Cloud Elastic Compute Service (ECS) インスタンス上の自己管理 Kafka クラスターを使用する場合は、ECS インスタンスの IP アドレスを入力します。 他のタイプの Kafka クラスターを使用する場合は、Kafka ブローカーのパブリック IP アドレスまたはドメイン名を入力します。 説明 Azure Event Hubs 統合のためのプライベートネットワークアクセスの設定方法
      トピック	データをインポートする Kafka トピック。コンマ (,) で区切って複数のトピックを指定できます。
      消費グループ	ApsaraMQ for Kafka を使用していて、無料で使用できるグループ機能を有効にしていない場合は、コンシューマーグループを選択する必要があります。コンシューマーグループの作成方法の詳細については、「コンシューマーグループの作成」をご参照ください。
      開始位置	データのインポートを開始する位置。 最早: トピックで利用可能な最初のデータレコードからインポートを開始します。 最新: トピック内の最新のデータレコードからインポートを開始します。
      データフォーマット	インポートするデータの形式。 シンプルモード: データが単一行フォーマットの場合は、シンプルモード を選択します。 JSON 文字列: データが JSON フォーマットの場合は、JSON 文字列 を選択します。この設定では、JSON データの最初のレイヤーをキーと値のペアに解析します。
      配列要素の解析	配列要素の解析 を有効にすると、Simple Log Service は、インポート前に JSON 配列を配列要素ごとに個別のログエントリに分割します。
      エンコード形式	インポートするデータのエンコード形式。UTF-8 と GBK がサポートされています。
      VPC ベースのインスタンス ID	ApsaraMQ for Kafka クラスターや ECS インスタンス上の自己管理クラスターなど、Kafka クラスターが Virtual Private Cloud (VPC) 内にある場合は、SLS が内部ネットワーク経由でデータを読み取れるように VPC ID を指定します。 内部ネットワークアクセスは、セキュリティとネットワークの安定性を向上させます。 重要 Kafka クラスターは、CIDR ブロック 100.104.0.0/16 からアクセス可能である必要があります。
      時間設定
      時間	ご利用の Kafka データ内でログ時間を表すフィールド。SLS はこのフィールドを、インポートされた各ログエントリのタイムスタンプとして使用します。
      時間フィールド抽出の正規表現	データフォーマット を シンプルモード に設定した場合、Kafka データから時間を抽出するための正規表現を指定する必要があります。 例えば、ログの内容が message with time 2022-08-08 14:20:20 の場合、時間フィールド抽出の正規表現を \d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d に設定できます。
      時刻フィールドの形式	時間フィールドの値を解析するための形式。 yyyy-MM-dd HH:mm:ss などの Java SimpleDateFormat パターン。詳細については、「Class SimpleDateFormat」をご参照ください。一般的な時間形式については、「時間形式」をご参照ください。 エポックタイム形式。有効な値は epoch、epochMillis、epochMacro、epochNano です。
      タイムゾーン	時間フィールドのタイムゾーン。 時間形式が epoch の場合、この設定は不要です。
      デフォルトの時間ソース	時間抽出が設定されていない、または失敗した場合のフォールバック時間ソース。オプションには、現在のシステム時間と Kafka メッセージのタイムスタンプが含まれます。
      詳細設定
      ログコンテキスト	ログコンテキスト を有効にすると、元の Kafka パーティション内で特定のログエントリに先行するまたは後続するログエントリを表示できます。
      通信プロトコル	パブリックネットワーク経由でインポートされるデータについては、ユーザー認証付きの暗号化された接続を使用することを推奨します。以下の例のように通信プロトコルを設定してください。 protocol フィールドは、plaintext、ssl、sasl_plaintext、および sasl_ssl をサポートしています。接続の暗号化とユーザー認証の両方が必要となる sasl_ssl に設定することを推奨します。 protocol を sasl_plaintext または sasl_ssl に設定した場合は、sasl ノードを設定する必要があります。mechanism フィールドは、ユーザー名とパスワードの認証メカニズムを表す PLAIN、SCRAM-SHA-256、または SCRAM-SHA-512 にすることができます。 { "protocol":"sasl_plaintext", "sasl":{ "mechanism":"PLAIN", "username":"xxx", "password":"yyy" } }
      プライベートドメイン名の解析	Alibaba Cloud ECS インスタンスにデプロイされた Kafka ブローカーが相互に通信するために内部ドメイン名を使用する場合、各ブローカーに対応するドメイン名と IP アドレスを指定する必要があります。以下のコードは一例です。 { "hostname#1":"192.168.XX.XX", "hostname#2":"192.168.XX.XX", "hostname#3":"192.168.XX.XX" }

   2. インポート結果を確認するには、プレビュー をクリックします。

   3. 結果を確認したら、次へ をクリックします。

5. インデックスを作成してデータをプレビューし、[次へ] をクリックします。デフォルトでは、SLS でフルテキストインデックスが有効になっています。収集されたログに対して手動でフィールドインデックスを作成するか、[インデックスの自動生成] をクリックすることもできます。クリックすると、SLS がフィールドインデックスを生成します。詳細については、「インデックスの作成 — 変更手順」をご参照ください。

   重要

   ログのすべてのフィールドをクエリする場合は、フルテキストインデックスの使用を推奨します。特定のフィールドのみをクエリする場合は、フィールドインデックスの使用を推奨します。これにより、インデックストラフィックを削減できます。フィールドを分析する場合は、フィールドインデックスを作成する必要があります。分析のためには、クエリ文に SELECT 文を含める必要があります。

6. ログ照会 をクリックして、クエリと分析ページに移動し、データインポートを検証します。

   約 1 分待機します。Kafka データが表示されれば、インポートは成功です。

データインポート設定の表示

コンソールでデータインポート設定の詳細と統計レポートを表示できます。

1. [プロジェクト] セクションで、目的のプロジェクトをクリックします。

2. 送信先の Logstore に移動し、データのインポート > データのインポート を選択して、構成名をクリックします。

3. インポート設定の概要 ページで、構成の基本情報と統計レポートを確認できます。

関連操作

インポート設定の概要 ページで、次の操作も実行できます。

• 設定の変更

  設定の変更 をクリックして構成設定を変更します。パラメーターの詳細については、「データインポート構成を作成する」をご参照ください。

• 設定の削除

  インポート構成を削除するには、設定の削除 をクリックします。

  警告

  この操作は元に戻せません。慎重に実行してください。

• タスクの停止

  停止 をクリックしてインポートタスクを停止します。

よくある質問

エラー処理メカニズム