Apache Spark StreamingのtKafkaInputAvroプロパティ

[Schema] (スキーマ)と[Edit schema] (スキーマを編集)

スキーマとは行の説明のことです。処理して次のコンポーネントに渡すフィールド(カラム)数を定義します。Sparkジョブを作成する場合、フィールドの命名時は予約語のlineを避けます。

[Broker list] (ブローカーリスト)

使用するKafkaクラスターのブローカーノードのアドレスを入力します。

このアドレスの形式はhostname:portです。この情報は、このKafkaクラスター内のホスティングノードの名前とポートです。

複数のアドレスを指定する必要がある場合は、コンマ(,)で区切ります。

[Starting offset] (開始オフセット)

消費されるトピックメッセージの開始ポイントを選択します。

Kafkaでは、メッセージのID連番はoffsetと呼ばれます。このリストから、トピック全体の最も古いメッセージから消費を開始するには[From beginning] (最初から)を選択し、同じコンシューマーグループによって消費され、オフセットがSparkチェックポイント内でSparkによって追跡される最新のメッセージから開始するには[From latest] (最新から)を選択します。

コンポーネントが消費されたメッセージの位置を記憶できるようにするには、ジョブの[Run] (実行)ビューの[Spark configuration] (Spark設定)タブでSpark Streamingのチェックポインティングをアクティブにする必要があります。

各コンシューマーグループには、消費したメッセージの位置を記憶するための独自のカウンターがあります。このため、コンシューマーグループが特定のトピックのメッセージの消費を開始すると、コンシューマグループはトピック全体ではなく、このグループが消費を停止する位置に関してのみ最新のメッセージを認識します。この原則に基づいて、次の動作が予想されます。

• トピックにたとえば100のメッセージがあるとします。コンシューマグループがオフセット50のメッセージで消費を停止した場合、[From latest] (最新から)を選択すると、同じコンシューマグループはオフセット 51から再開します。

• 新しいコンシューマーグループを作成するか、既存のコンシューマーグループをリセットした場合(どちらの場合も、このグループがこのトピックのメッセージを消費していないことを意味します)、最新のグループから開始すると、この新しいグループが開始し、オフセット101を待機します。

tKafkaInputがメッセージフィードを受け取るトピック名を入力します。

[Group ID] (グループID)

現在のコンシューマー(tKafkaInputコンポーネント)が属するようにしたいコンシューマーグループの名前を入力します。

この時点でこのコンシューマーグループが存在していない場合、実行時に作成されます。

このプロパティは、Spark 2.0を使っている場合、または使うHadoopディストリビューションがSpark 2.0を実行している場合のみ利用できます。使っているSparkのバージョンが不明な場合は、クラスターの管理者に詳細をお問い合わせください。

[Set number of records per second to read from each Kafka partition] (各Kafkaパーティションから読み取る秒当たりのレコード件数を設定)

この数値を二重引用符で囲んで入力し、処理のために送信される各バッチのサイズを制限します。

たとえば、100を入力し、Spark設定タブで定義したバッチ値が2秒の場合、各バッチのパーティションからのサイズは200メッセージです。

このチェックボックスをオフにした場合、コンポーネントは、1秒間で単一のバッチに全メッセージを読み取ろうとしますが、大量のメッセージのためにジョブが応答しなくなる可能性があります。

[Use SSL/TLS] (SSL/TLSを使用)

SSLまたはTLS暗号化接続を有効にする場合は、このチェックボックスを選択します。

次に、同じジョブ内のtSetKeyStoreコンポーネントを使用して暗号化情報を指定する必要があります。

このプロパティは、Spark 2.0を使っている場合、または使うHadoopディストリビューションがSpark 2.0を実行している場合のみ利用できます。使っているSparkのバージョンが不明な場合は、クラスターの管理者に詳細をお問い合わせください。

TrustStoreファイルと使用されているKeyStoreファイルは、Sparkエグゼキューターをホスティングしているすべての単一のSparkノードにローカルに保管されている必要があります。

[Use Kerberos authentication] (Kerberos認証を使用)

使用するKafkaクラスターをKerberosで保護する場合は、このチェックボックスを選択して、定義する関連パラメーターを表示します。

• JAAS configuration path (JAAS設定パス): パスを入力するか、Kafkaへのクライアントの認証を実行するジョブによって使われるJAAS設定ファイルを参照します。

  このJAASファイルには、kinitモードまたはkeytabモードのいずれかを使用して、クライアント(TalendによるKafka関連ジョブ)をKafkaブローカーノードに接続する方法が指定されています。JAASファイルは、これらのジョブが実行されるマシンに保管されている必要があります。

  Talend、Kerberos、またはKafkaはこのJAASファイルを提供していません。ユーザーの組織のセキュリティ戦略に応じて、Configuring Kafka clientの説明に従ってJAASファイルを作成する必要があります。

• Kafka brokers principal name (Kafkaブローカーのプリンシパル名): ブローカークラスターの作成時にブローカーのために定義したKerberosのプリンシパルの主要部分を入力します。たとえばkafka/kafka1.hostname.com@EXAMPLE.COMというプリンシパルの場合、このフィールドへの入力に使用する主要部分はkafkaになります。

• Set kinit command path (kinitコマンドパス): Kerberosは、そのkinit実行可能ファイルへのデフォルトのパスを使用します。このパスを変更した場合は、このチェックボックスを選択してカスタムアクセスパスを入力します。

  このチェックボックスをオフにする場合は、デフォルトのパスが使用されます。

• Set Kerberos configuration path (Kerberos設定パス): Kerberosは、たとえば、Kerberos 5の設定ファイルであるkrb5.confファイル(または、Windowsの場合はkrb5.ini)へのデフォルトのパスを使用します。このパスを変更した場合は、このチェックボックスを選択して、Kerberos設定ファイルへのカスタムアクセスパスを入力します。

  このチェックボックスをオフにすると、必要な設定情報を見つけるために、Kerberosによって所定の戦略が適用されます。この戦略の詳細は、Kerberos requirementsでLocating the krb5.conf Configuration Fileをご覧ください。

KafkaクラスターをKerberosで保護する方法は、SASLを使用した認証をご覧ください。

このチェックボックスはKafka 0.9.0.1以降で使えます。

[Use Schema Registry] (スキーマレジストリーを使用)

スキーマレジストリーの詳細は、Confluentのドキュメンテーションをご覧ください。

このオプションは、Talendが提供する8.0.1-R2022-12以降のTalend Studioマンスリーアップデートをインストール済みである場合に利用できます。詳細は管理者にお問い合わせください。

[Kafka properties] (Kafkaのプロパティ)

カスタマイズする必要があるKafkaコンシューマープロパティをこのテーブルに追加します。たとえば、ZkTimeoutExceptionを避けるために特定のzookeeper.connection.timeout.ms値を設定できます。

このテーブルで定義できるコンシューマプロパティの詳細は、次のWebサイトにあるKafkaのドキュメンテーションでコンシューマ設定について説明しているセクション(http://kafka.apache.org/documentation.html#consumerconfigs)をご覧ください。

[Use hierarchical mode] (階層モードを使用)

バイナリ(階層を含む) Avroスキーマを、現在のコンポーネントのスキーマエディターに定義されているフラットスキーマにマップする場合は、このこのチェックボックスを選択します。処理するAvroメッセージto be processed is がフラットの場合は、このチェックボックスをオフのままにしておきます。

オンにする場合は、次のパラメーターを設定する必要があります。

• [Local path to the avro schema] (avroスキーマへのローカルパス): 処理するAvroデータのスキーマを定義するファイルに移動します。

• [Mapping]: 現在のコンポーネントのスキーマカラムと処理する階層Avroメッセージに保存されているデータの間にマップを作成します。[Node] (ノード)カラム内に、Avroメッセージから読み取るデータをポイントするJSONパスを入力する必要があります。

使用ルール

このコンポーネントは、開始コンポーネントとして使用され、出力リンクを必要とします。

このコンポーネントは、所属するSpark Streamingのコンポーネントのパレットと共に、Spark Streamingジョブを作成している場合にだけ表示されます。

特に明記していない限り、このドキュメンテーションのシナリオでは、標準ジョブ、つまり従来の Talend Data Integrationジョブだけを扱います。

Sparkの現在のコンポーネントの実装では、KafkaオフセットはSpark自体によって自動的に管理されます。つまり、ZooKeeperまたはKafkaにコミットされる代わりに、オフセットはSparkチェックポイント内で追跡されます。この実装の詳細は、Sparkのドキュメンテーションでダイレクトアプローチのセクションをご覧ください。