1. Home
  2. Cloud
  3. Confluent Cloud への外部システムの接続

重要

このページの日本語コンテンツは古くなっている可能性があります。最新の英語版コンテンツをご覧になるには、こちらをクリックしてください。

Splunk Sink Connector for Confluent Cloud

注釈

Confluent Platform 用にコネクターをローカルにインストールする場合は、「Splunk Sink Connector for Confluent Platform」を参照してください。

Splunk Sink Connector は、Splunk HTTP Event Collector（HEC） を使用して、メッセージを Apache Kafka® から Splunk に移動するために使用します。

機能

Splunk Sink Connector は、以下の機能をサポートしています。

  • 少なくとも 1 回のデリバリー: コネクターによって、Kafka のトピックからのレコードが少なくとも 1 回は配信されることが保証されます。
  • 複数のタスクのサポート: 1 つまたは複数のタスクの実行をサポートしています。タスクが多いほどパフォーマンスが向上する可能性があります（複数のタスクを実行するとコンシューマーラグが減少します）。

Confluent Cloud API for Connect の詳細と使用例については、「Confluent Cloud API for Connect」セクションを参照してください。

制限

以下の情報を確認してください。

クイックスタート

このクイックスタートを使用して、Confluent Cloud Splunk Sink Connector の利用を開始することができます。このクイックスタートでは、コネクターを選択し、Splunk にイベントをストリーミングするようにコネクターを構成するための基本的な方法について説明します。

前提条件
  • アマゾンウェブサービス （ AWS ）、Microsoft Azure （ Azure ）、または Google Cloud Platform （ GCP ）上の Confluent Cloud クラスターへのアクセスを許可されていること。
  • Confluent CLI がインストールされ、クラスター用に構成されていること。「Confluent CLI のインストール」を参照してください。
  • Splunk へのアクセスを許可されていること。
  • スキーマレジストリ ベースのフォーマット（Avro、JSON_SR（JSON スキーマ）、Protobuf など）を使用するには、Schema Registry を有効にしておく必要があります。詳細については、「スキーマレジストリ Enabled Environments」を参照してください。
  • シンクコネクターを作成する前に、Confluent Cloud クラスター上にソース Kafka トピックが 1 つ以上存在している必要があります。

Confluent Cloud Console の使用

ステップ 1: Confluent Cloud クラスターを起動します。

インストール手順については、「Quick Start for Confluent Cloud」を参照してください。

ステップ 2: コネクターを追加します。

左のナビゲーションメニューの Data integration をクリックし、Connectors をクリックします。クラスター内に既にコネクターがある場合は、+ Add connector をクリックします。

ステップ 3: コネクターを選択します。

Click the Splunk Sink connector card.

Splunk Sink Connector Card

ステップ 4: コネクターの詳細情報を入力します。

注釈

  • すべての 前提条件 を満たしていることを確認します。
  • アスタリスク（ * ）は必須項目であることを示しています。
  • 以下の手順では、オプションの UI プロパティについては説明していません。構成プロパティの値と説明については、「構成プロパティ」を参照してください。

Add Splunk Sink Connector 画面で、以下を実行します。

既に Kafka トピックを用意している場合は、Topics リストから接続するトピックを選択します。

新しいトピックを作成するには、+Add new topic をクリックします。

ステップ 5: レコードを確認します。

レコードが Splunk に生成されていることを確認します。

Confluent Cloud API for Connect の詳細と使用例については、「Confluent Cloud API for Connect」セクションを参照してください。

ちなみに

コネクターを起動すると、デッドレターキューのトピックが自動的に作成されます。詳細については、「Confluent Cloud デッドレターキュー」を参照してください。

Confluent CLI の使用

Confluent CLI でコネクターをセットアップして実行するには、次の手順を実行します。

注釈

ステップ 1: 使用可能なコネクターをリスト表示します。

以下のコマンドを入力して、使用可能なコネクターをリスト表示します。

confluent connect plugin list

ステップ 2: コネクターの必須の構成プロパティを表示します。

以下のコマンドを実行して、コネクターの必須プロパティを表示します。

confluent connect plugin describe <connector-catalog-name>

例:

confluent connect plugin describe SplunkSink

出力例:

Following are the required configs:
connector.class: SplunkSink
topics
input.data.format
name
kafka.api.key
kafka.api.secret
splunk.hec.uri
splunk.hec.token
tasks.max

ステップ 3: コネクターの構成ファイルを作成します。

コネクター構成プロパティを含む JSON ファイルを作成します。以下の例は、コネクターの必須プロパティを示しています。

{
  "connector.class": "SplunkSink",
  "topics": "orders",
  "name": "SplunkSinkConnector_0",
  "input.data.format": "AVRO",
  "kafka.auth.mode": "KAFKA_API_KEY",
  "kafka.api.key": "<my-kafka-api-key>",
  "kafka.api.secret": "<my-kafka-api-secret>",
  "splunk.hec.uri": "https://hec1.splunk.com:8088,https://hec2.splunk.com:8088,https://hec3.splunk.com:8088",
  "splunk.hec.token": "<token>",
  "tasks.max": "1",

}

以下のプロパティ定義に注意してください。

  • "connector.class": コネクターのプラグイン名を指定します。
  • "input.data.format": Kafka 入力レコード値のフォーマット（Kafka トピックから送られるデータ）を設定します。指定可能なエントリは、AVRO、JSON_SR、PROTOBUF、JSON、または STRING です。スキーマベースのメッセージフォーマット（たとえば、Avro、JSON_SR（JSON スキーマ）、および Protobuf）を使用するには、Confluent Cloud Schema Registry を構成しておく必要があります。
  • "name": 新しいコネクターの名前を設定します。

  • "kafka.auth.mode": 使用するコネクターの認証モードを指定します。オプションは SERVICE_ACCOUNT または KAFKA_API_KEY （デフォルト）です。API キーとシークレットを使用するには、構成プロパティ kafka.api.key と kafka.api.secret を構成例（前述）のように指定します。サービスアカウント を使用するには、プロパティ kafka.service.account.id=<service-account-resource-ID> に リソース ID を指定します。使用できるサービスアカウントのリソース ID のリストを表示するには、次のコマンドを使用します。

    confluent iam service-account list

    例:

    confluent iam service-account list

   Id     | Resource ID |       Name        |    Description
+---------+-------------+-------------------+-------------------
   123456 | sa-l1r23m   | sa-1              | Service account 1
   789101 | sa-l4d56p   | sa-2              | Service account 2

  • "splunk.hec.uri": すべての Splunk インデクサーの FQDN または IP アドレスのコンマ区切りのリストを追加するか、ロードバランサー を追加します。Splunk インデクサーでは、ロードバランシングにラウンドロビン方式のスケジューリングが使用されます。たとえば、https://hec1.splunk.com:8088,https://hec2.splunk.com:8088,https://hec3.splunk.com:8088 のようになります。

  • "splunk.hec.token": Splunk HTTP イベントコレクタトークン を追加します。

  • "tasks.max": このコネクターで使用できる タスク の最大数を入力します。タスクが多いほどパフォーマンスが向上する可能性があります。

  • "topics": 特定のトピック名を指定するか、複数のトピック名をコンマ区切りにしたリストを指定します。

Single Message Transforms: CLI を使用した SMT の追加の詳細については、Single Message Transforms（SMT） のドキュメントを参照してください。

すべてのプロパティの値と説明については、「構成プロパティ」を参照してください。

ステップ 3: プロパティファイルを読み込み、コネクターを作成します。

以下のコマンドを入力して、構成を読み込み、コネクターを起動します。

confluent connect create --config <file-name>.json

例:

confluent connect create --config splunk-sink-config.json

出力例:

Created connector SplunkSinkConnector_0 lcc-do6vzd

ステップ 4: コネクターのステータスを確認します。

以下のコマンドを入力して、コネクターのステータスを確認します。

confluent connect list

出力例:

ID           |             Name                | Status  | Type | Trace
+------------+---------------------------------+---------+------+-------+
lcc-do6vzd   | SplunkSinkConnector_0           | RUNNING | sink |       |

ステップ 5: レコードを確認します。

レコードが Splunk に取り込まれていることを確認します。

Confluent Cloud API for Connect の詳細と使用例については、「Confluent Cloud API for Connect」セクションを参照してください。

ちなみに

コネクターを起動すると、デッドレターキューのトピックが自動的に作成されます。詳細については、「Confluent Cloud デッドレターキュー」を参照してください。

構成プロパティ

このコネクターでは、以下のコネクター構成プロパティを使用します。

データの取得元とするトピック（Which topics do you want to get data from?）

topics

特定のトピック名を指定するか、複数のトピック名をコンマ区切りにしたリストを指定します。

  • 型: list
  • 重要度: 高

入力メッセージ（Input messages）

input.data.format

Kafka 入力レコード値のフォーマットを設定します。指定可能なエントリは、AVRO、JSON、JSON_SR、PROTOBUF、または STRING です。スキーマベースのメッセージフォーマット（AVRO、JSON_SR、PROTOBUF など）を使用する場合は、Confluent Cloud Schema Registry を構成しておく必要がある点に注意してください。

  • 型: string
  • 重要度: 高

データへの接続方法（How should we connect to your data?）

name

コネクターの名前を設定します。

  • 型: string
  • 指定可能な値: 最大 64 文字の文字列
  • 重要度: 高

Kafka クラスターの認証情報（Kafka Cluster credentials）

kafka.auth.mode

Kafka の認証モード。KAFKA_API_KEY または SERVICE_ACCOUNT を指定できます。デフォルトは KAFKA_API_KEY モードです。

  • 型: string
  • デフォルト: KAFKA_API_KEY
  • 指定可能な値: KAFKA_API_KEY、SERVICE_ACCOUNT
  • 重要度: 高
kafka.api.key
  • 型: password
  • 重要度: 高
kafka.service.account.id

Kafka クラスターとの通信用の API キーを生成するために使用されるサービスアカウント。

  • 型: string
  • 重要度: 高
kafka.api.secret
  • 型: password
  • 重要度: 高

Splunk への接続方法（How should we connect to Splunk?）

splunk.hec.uri

すべての Splunk インデクサーの FQDN または IP のコンマ（「,」）区切りのリスト、またはロードバランサー。コネクターでは、ラウンドロビン方式でインデクサーに負荷が分散されます。たとえば、https://hec1.splunk.com:8088,https://hec2.splunk.com:8088,https://hec3.splunk.com:8088 のようになります。

  • 型: string
  • 重要度: 高
splunk.hec.token

Splunk HTTP イベントコレクタトークン。

  • 型: password
  • 重要度: 高
splunk.hec.ssl.validate.certs

HTTPS 証明書の検証を有効または無効にします。

  • 型: boolean
  • デフォルト: true
  • 重要度: 中
splunk.hec.ssl.trust.store.file

SSL 接続の検証に要求される証明書を含む証明書トラストストア。

  • 型: password
  • デフォルト: [hidden]
  • 重要度: 高
splunk.hec.ssl.trust.store.password

証明書トラストストアのパスワード。

  • 型: password
  • 重要度: 高

メタデータの構成（Metadata configuration）

splunk.indexes

複数のトピックをインデクサーに設定する場合に、Kafka のトピックのデータに対応する Splunk のインデックス名をコンマ区切りで指定します（"prod-index1,prod-index2,prod-index3"）。

  • 型: string
  • デフォルト: default
  • 重要度: 中
splunk.sourcetypes

Kafka のトピックのデータに対応する Splunk イベント sourcetype のメタデータ。

  • 型: string
  • 重要度: 中
splunk.sources

Kafka のトピックのデータに対応する Splunk イベントソースのメタデータ。

  • 型: string
  • 重要度: 中

エンドポイントの構成（Endpoint configuration）

splunk.hec.raw

true に設定されている場合、コネクターは /raw HEC エンドポイントを使用してデータを取り込みます。

  • 型: boolean
  • デフォルト: false
  • 重要度: 中
splunk.hec.raw.line.breaker

/raw HEC エンドポイントにのみ適用されます。設定は、Splunk でイベントを正しく区切るためのカスタムのラインブレーカーの指定に使用されます。

  • 型: string
  • 重要度: 中
splunk.hec.json.event.enrichment

/event HEC エンドポイントにのみ適用されます。この設定は、未加工データにメタデータフィールドを追加して補強するために使用されます。"," で区切られたキーと値のペアのリストで指定します。

  • 型: string
  • 重要度: 低
splunk.hec.track.data

/event HEC エンドポイントにのみ適用されます。true に設定すると、データ損失とデータ取り込みレイテンシのメタデータが未加工データとともにインデックス化されます。

  • 型: boolean
  • デフォルト: false
  • 重要度: 低

HEC の構成（HEC configuration）

splunk.hec.http.keepalive

HTTP 接続のキープアライブを有効または無効にします。

  • 型: boolean
  • デフォルト: true
  • 重要度: 中
splunk.hec.max.http.connection.per.channel

Splunk へのイベント送信時に、1 つの HEC チャネルにプールされる最大 HTTP 接続数。

  • 型: int
  • デフォルト: 2
  • 重要度: 中
splunk.hec.total.channels

Splunk へのイベントの送信に使用される HEC チャネルの総数。

  • 型: int
  • デフォルト: 2
  • 重要度: 高
splunk.hec.socket.timeout

ネットワークでデータの読み取りおよび書き込みを行う最長時間（秒）。これを過ぎると内部 TCP ソケットがタイムアウトします。

  • 型: int
  • デフォルト: 60
  • 重要度: 低
splunk.hec.use.record.timestamp

true に設定すると、タイムスタンプが Kafka レコードから取得され、Splunk に HEC メタデータのオーバーライドとして渡されます。

  • 型: boolean
  • デフォルト: true
  • 重要度: 中
splunk.hec.threads

1 つのコネクタータスクで HEC を介してデータ取り込みを行うために生成されるスレッドの数。

  • 型: int
  • デフォルト: 1
  • 指定可能な値: [1,...,10]
  • 重要度: 低
splunk.hec.max.outstanding.events

コネクターでメモリーに保持できる、確認応答されていないイベントの最大数。確認応答されていないイベントがこの値を超えると、バックプレッシャーイベントをトリガーし、収集速度が低下します。

  • 型: int
  • デフォルト: 10000
  • 指定可能な値: [10000,...,100000]
  • 重要度: 中
splunk.hec.max.retries

失敗したバッチの再試行の最大回数。これを超えて再試行は行われません。デフォルトは -1 に設定されており、無限に再試行が行われます。

  • 型: int
  • デフォルト: -1
  • 重要度: 中
splunk.hec.backoff.threshhold.seconds

Splunk へのイベント送信でエラーが発生した場合に、再送信を試行する前にコネクターが待機する時間。

  • 型: int
  • デフォルト: 60
  • 重要度: 中
splunk.hec.json.event.formatted

既に HEC フォーマットであるイベントについては、true に設定します。

  • 型: boolean
  • デフォルト: false
  • 重要度: 低
splunk.hec.max.batch.size

Splunk にイベントを送信する際の最大バッチサイズ。このサイズは、バイトサイズではなく実際の Kafka イベントの数です。

  • 型: int
  • デフォルト: 500
  • 重要度: 中
splunk.hec.lb.poll.interval

この設定は、ロードバランサーのポーリング間隔を制御します。

  • 型: int
  • デフォルト: 120
  • 重要度: 低
splunk.flush.window

kafka から Splunk にイベントがフラッシュされる間隔（秒）。

  • 型: int
  • デフォルト: 30
  • 重要度: 低

確認応答の構成（Acknowledgement configuration）

splunk.hec.ack.enabled

true に設定すると、コネクターにより、Kafka オフセットのチェックポイント設定前に、POST イベントに対して ACK がポーリングされます。この設定によってデリバリーの保証が実装されるため、データ損失の防止に使用されます。

  • 型: boolean
  • デフォルト: false
  • 重要度: 中
splunk.hec.ack.poll.interval

この設定は、splunk.hec.ack.enabled が true に設定されている場合にのみ適用されます。内部的には、イベントの ACK のポーリング間隔を制御します。

  • 型: int
  • デフォルト: 10
  • 重要度: 中
splunk.hec.ack.poll.threads

この設定は、splunk.hec.ack.enabled が true に設定されている場合にのみ適用されます。この設定は、イベントの ACK のポーリングのために生成する必要があるスレッド数を制御します。

  • 型: int
  • デフォルト: 1
  • 指定可能な値: [1,...,10]
  • 重要度: 中
splunk.hec.event.timeout

この設定は、splunk.hec.ack.enabled が true に設定されている場合にのみ適用されます。イベントが Splunk に送信されてから ACK されていない場合に、この設定の期間に応じて、コネクターはタイムアウトして再送信することを待機します。

  • 型: int
  • デフォルト: 300
  • 重要度: 中

ヘッダーの構成（Headers configuration）

splunk.header.support

true に設定すると、コネクターは Splunk イベントのメタデータとして使用するために Kafka のヘッダーを解析します。

  • 型: boolean
  • デフォルト: false
  • 重要度: 中
splunk.header.custom

この設定では、これらの値を持つ kafka レコードヘッダーが検索され、存在する場合は各イベントに追加されます。カスタムヘッダーは、ヘッダーが複数の場合、コンマで区切って構成します。たとえば、"custom_header_1,custom_header_2,custom_header_3" のように設定します。

  • 型: string
  • 重要度: 中
splunk.header.index

Splunk ヘッダーの Index に使用するヘッダー。

  • 型: string
  • デフォルト: splunk.header.index
  • 重要度: 中
splunk.header.source

Splunk ヘッダーの Source に使用するヘッダー。

  • 型: string
  • デフォルト: splunk.header.source
  • 重要度: 中
splunk.header.sourcetype

Splunk ヘッダーの Sourcetype に使用するヘッダー。

  • 型: string
  • デフォルト: splunk.header.sourcetype
  • 重要度: 中
splunk.header.host

Splunk ヘッダーの Host に使用するヘッダー。

  • 型: string
  • デフォルト: splunk.header.host
  • 重要度: 中

このコネクターのタスク数（Number of tasks for this connector）

tasks.max
  • 型: int
  • 指定可能な値: [1,...]
  • 重要度: 高

次のステップ

参考

フルマネージド型の Confluent Cloud コネクターが Confluent Cloud ksqlDB でどのように動作するかを示す例については、「Cloud ETL のデモ」を参照してください。この例では、Confluent CLI を使用して Confluent Cloud のリソースを管理する方法についても説明しています。

../_images/topology.ja.png