Apache Spark StreamingのtKafkaInputAvroプロパティ - 7.3

Kafka

EnrichVersion
Cloud
7.3
EnrichProdName
Talend Big Data
Talend Big Data Platform
Talend Data Fabric
Talend Open Studio for Big Data
Talend Real-Time Big Data Platform
EnrichPlatform
Talend Studio
task
ジョブデザインと開発 > サードパーティーシステム > メッセージングコンポーネント > Kafka
データガバナンス > サードパーティーシステム > メッセージングコンポーネント > Kafka
データクオリティとプレパレーション > サードパーティーシステム > メッセージングコンポーネント > Kafka

このプロパティは Spark Streamingジョブフレームワークで実行されているtKafkaInputAvroを設定するために使います。

Spark Streaming tKafkaInputAvroコンポーネントはメッセージングファミリーのコンポーネントです。

このコンポーネントは、Talend Real Time Big Data PlatformおよびTalend Data Fabricで使用できます。

基本設定

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

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

ブローカーリスト

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

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

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

開始オフセット

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

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

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

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

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

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

トピック名

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

グループID

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

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

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

各Kafaパーティションから読み込む秒当たりのレコード件数を設定

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

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

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

SSL/TLSの使用

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

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

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

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

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以降で使えます。

詳細設定

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パスを入力する必要があります。

使用方法

[Usage rule] (使用ルール)

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

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

特に明記していない限り、このドキュメントのシナリオでは、[Standard] (標準)ジョブ、つまり従来の Talend データ統合ジョブだけを扱います。

Sparkの現在のコンポーネントの実装では、KafkaオフセットはSpark自体によって自動的に管理されます。つまり、ZookeeperまたはKafkaにコミットされる代わりに、オフセットはSparkチェックポイント内で追跡されます。この実装については、Sparkドキュメンテーションでダイレクトアプローチのセクションを参照してください: http://spark.apache.org/docs/latest/streaming-kafka-integration.html