# tKafkaInputAvro

処理する必要のあるAvro形式のメッセージを、設計しているジョブ内の後続コンポーネントに送信します。

tKafkaInputAvroは、これらのメッセージに対して変換を実行するジョブにAVRO形式のメッセージを送信するジェネリックのメッセージブローカーです。このコンポーネントは、avro-toolsライブラリーによって作成されたAVROメッセージを処理できません。

デフォルトで、このコンポーネントはTalend Studioと共には出荷されていません。機能マネージャーを使ってインストールする必要があります。 詳細は、[機能マネージャーを使って機能をインストール](https://help.qlik.com/talend/r/ja-JP/8.0/studio-user-guide/install-features-to-talend-studio)をご覧ください。

## Apache Spark StreamingのtKafkaInputAvroプロパティ

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

Spark StreamingのtKafkaInputAvroコンポーネントは、メッセージングファミリーに属しています。

このコンポーネントはTalend Real-Time Big Data PlatformとTalend Data Fabricで利用できます。

情報メモ重要: Spark Streamingジョブの場合、Talend StudioはKafkaの特定のバージョンをサポートしておらず、Sparkが提供するKafkaブローカーのバージョン互換性に依存しています。サポートされているKafkaブローカーのバージョンは、使用するSparkのバージョンによって異なります。Sparkの各バージョンについて、Talend Studioは、Sparkが提供するKafkaブローカーのターゲットバージョンをサポートしています。現在のところ、Talend StudioはSparkとの互換性声明に依存しているため、Kafkaブローカーのバージョン0.10.0以降をサポートしています。詳細は、Sparkの公式ドキュメンテーションで[Spark Streaming + Kafka Integration Guide](https://spark.apache.org/docs/latest/streaming-kafka-integration.html)をご覧ください。

## 基本設定

| [Properties] (プロパティ) | 説明 |
| --- | --- |
| [Schema] (スキーマ)と[Edit schema] (スキーマを編集) | スキーマとは行の説明のことです。処理して次のコンポーネントに渡すフィールド(カラム)数を定義します。Sparkジョブを作成する場合、フィールドの命名時は予約語のlineを避けます。 |
| [Broker list] (ブローカーリスト) | 使用するKafkaクラスターのブローカーノードのアドレスを入力します。<br><br>このアドレスの形式は*hostname:port*です。この情報は、このKafkaクラスター内のホスティングノードの名前とポートです。<br><br>複数のアドレスを指定する必要がある場合は、コンマ(*,*)で区切ります。 |
| [Starting offset] (開始オフセット) | 消費されるトピックメッセージの開始ポイントを選択します。<br><br>Kafkaでは、メッセージのID連番は*offset*と呼ばれます。このリストから、トピック全体の最も古いメッセージから消費を開始するには[From beginning] (最初から)を選択し、同じコンシューマーグループによって消費され、オフセットがSparkチェックポイント内でSparkによって追跡される最新のメッセージから開始するには[From latest] (最新から)を選択します。<br><br>コンポーネントが消費されたメッセージの位置を記憶できるようにするには、ジョブの[Run] (実行)ビューの[Spark configuration] (Spark設定)タブでSpark Streamingのチェックポインティングをアクティブにする必要があります。<br><br>各コンシューマーグループには、消費したメッセージの位置を記憶するための独自のカウンターがあります。このため、コンシューマーグループが特定のトピックのメッセージの消費を開始すると、コンシューマーグループはトピック全体ではなく、このグループが消費を停止する位置に関してのみ最新のメッセージを認識します。この原則に基づいて、次の動作が予想されます。<br><br><ul> <li> <p>トピックにたとえば<em>100</em>のメッセージがあるとします。コンシューマーグループがオフセット<em>50</em>のメッセージで消費を停止した場合、<span>[From latest] (最新から)</span>をオンにすると、同じコンシューマーグループはオフセット <em>51</em>から再開します。</p> </li> <li> <p>新しいコンシューマーグループを作成するか、既存のコンシューマーグループをリセットした場合(どちらの場合も、このグループがこのトピックのメッセージを消費していないことを意味します)、最新のグループから開始すると、この新しいグループが開始し、オフセット<em>101</em>を待機します。</p> </li> </ul> |
| [Topic name] (トピック名) | tKafkaInputがメッセージフィードを受け取るトピック名を入力します。 |
| [Group ID] (グループID) | 現在のコンシューマー(tKafkaInputコンポーネント)が属するようにしたいコンシューマーグループの名前を入力します。<br><br>この時点でこのコンシューマーグループが存在していない場合、実行時に作成されます。<br><br>このプロパティは、Spark 2.0を使っている場合、または使うHadoopディストリビューションがSpark 2.0を実行している場合のみ利用できます。使っているSparkのバージョンが不明な場合は、クラスターの管理者に詳細をお問い合わせください。 |
| [Set number of records per second to read from each Kafka partition] (各Kafkaパーティションから読み取る秒当たりのレコード件数を設定) | この数値を二重引用符で囲んで入力し、処理のために送信される各バッチのサイズを制限します。<br><br>たとえば、*100*を入力し、Spark設定タブで定義したバッチ値が*2*秒の場合、各バッチのパーティションからのサイズは*200*メッセージです。<br><br>このチェックボックスをオフにした場合、コンポーネントは、1秒間で単一のバッチに全メッセージを読み取ろうとしますが、大量のメッセージのためにジョブが応答しなくなる可能性があります。 |
| [Use SSL/TLS] (SSL/TLSを使用) | SSLまたはTLS暗号化接続を有効にする場合は、このチェックボックスをオンにします。<br><br>次に、同じジョブ内のtSetKeyStoreコンポーネントを使用して暗号化情報を指定する必要があります。<br><br>このプロパティは、Spark 2.0を使っている場合、または使うHadoopディストリビューションがSpark 2.0を実行している場合のみ利用できます。使っているSparkのバージョンが不明な場合は、クラスターの管理者に詳細をお問い合わせください。<br><br>TrustStoreファイルと使用されているKeyStoreファイルは、Sparkエグゼキューターをホスティングしているすべての単一のSparkノードにローカルに保管されている必要があります。 |
| [Use Kerberos authentication] (Kerberos認証を使用) | 使用するKafkaクラスターをKerberosで保護する場合は、このチェックボックスをオンにして、定義する関連パラメーターを表示します。<br><br><ul> <li> <p> <span>JAAS configuration path (JAAS設定パス)</span>: パスを入力するか、Kafkaへのクライアントの認証を実行するジョブによって使われるJAAS設定ファイルを参照します。</p> <p>このJAASファイルには、kinitモードまたはkeytabモードのいずれかを使用して、クライアント(<em><span>Talend</span></em>によるKafka関連ジョブ)をKafkaブローカーノードに接続する方法が指定されています。JAASファイルは、これらのジョブが実行されるマシンに保管されている必要があります。</p> <p> <em><span>Talend</span></em>、Kerberos、またはKafkaはこのJAASファイルを提供していません。ユーザーの組織のセキュリティ戦略に応じて、<a href="http://kafka.apache.org/documentation.html#security_sasl_clientconfig" target="_blank">Configuring Kafka client</a>の説明に従ってJAASファイルを作成する必要があります。</p> </li> <li> <p> <span>Kafka brokers principal name (Kafkaブローカーのプリンシパル名)</span>: ブローカークラスターの作成時にブローカーのために定義したKerberosのプリンシパルの主要部分を入力します。たとえば<em>kafka/kafka1.hostname.com@EXAMPLE.COM</em>というプリンシパルの場合、このフィールドへの入力に使用する主要部分は<em>kafka</em>になります。</p> </li> <li> <p> <span>Set kinit command path (kinitコマンドパス)</span>: Kerberosは、そのkinit実行可能ファイルへのデフォルトのパスを使用します。このパスを変更した場合は、このチェックボックスをオンにしてカスタムアクセスパスを入力します。</p> <p>このチェックボックスをオフにする場合は、デフォルトのパスが使用されます。</p> </li> <li> <p> <span>Set Kerberos configuration path (Kerberos設定パス)</span>: Kerberosは、たとえば、Kerberos 5の設定ファイルである<em>krb5.conf</em>ファイル(または、Windowsの場合は<em>krb5.ini</em>)へのデフォルトのパスを使用します。このパスを変更した場合は、このチェックボックスをオンにして、Kerberos設定ファイルへのカスタムアクセスパスを入力します。</p> <p>このチェックボックスをオフにすると、必要な設定情報を見つけるために、Kerberosによって所定の戦略が適用されます。この戦略の詳細は、<a href="http://docs.oracle.com/javase/7/docs/technotes/guides/security/jgss/tutorials/KerberosReq.html" target="_blank">Kerberos requirements</a>で<em>Locating the krb5.conf Configuration File</em>をご覧ください。</p> </li> </ul><br><br>KafkaクラスターをKerberosで保護する方法は、[SASLを使用した認証](http://kafka.apache.org/documentation.html#security_sasl)をご覧ください。<br><br>このチェックボックスはKafka 0.9.0.1以降で使えます。 |
| [Use Schema Registry] (スキーマレジストリーを使用) | このチェックボックスをオンにすると、Confluent Schema Registryが使用され、定義する次の関連パラメーターが表示されます。<br><br><ul> <li> <span>URL</span>: スキーマレジストリインスタンスURLを入力します。</li> <li> <span>[Basic authentication] (基本認証)</span>: このチェックボックスをオンにして<span>[Username] (ユーザー名)</span>と<span>[Password] (パスワード)</span>に認証情報を入力します。</li> <li> <span>[Use the keystore of Kafka broker] (Kafkaブローカーのキーストアを使用)</span>: Kafkaブローカーで使われるものと同じtSetKeystoreコンポーネントを使ってSSLまたはTLSによる暗号化接続を有効にする場合は、このチェックボックスをオンにします。このチェックボックスは、<span>[Use SSL/TLS] (SSL/TLSを使用)</span>チェックボックスをオンにし、<span>[Set schema registry keystore] (スキーマレジストリーのキーストアを設定)</span>をオフにすると使用できます。</li> <li> <span>[Set schema registry keystore] (スキーマレジストリーのキーストアを設定)</span>: このチェックボックスをオンにすると、SSLまたはTLS暗号化接続が有効になります。次に、同じジョブ内の<span>tSetKeystore</span>コンポーネントを使用して暗号化情報を指定する必要があります。</li> </ul><br><br>スキーマレジストリーの詳細は、[Confluentのドキュメンテーション](https://docs.confluent.io/platform/current/schema-registry/index.html)をご覧ください。<br><br>このオプションは、Talendが提供する8.0.1-R2022-12以降のTalend Studioマンスリーアップデートをインストール済みである場合に利用できます。詳細は管理者にお問い合わせください。 |

## 詳細設定

| [Properties] (プロパティ) | 説明 |
| --- | --- |
| [Kafka properties] (Kafkaのプロパティ) | カスタマイズする必要があるKafkaコンシューマープロパティをこのテーブルに追加します。たとえば、*ZkTimeoutException*を避けるために特定の*zookeeper.connection.timeout.ms*値を設定できます。<br><br>このテーブルで定義できるコンシューマープロパティの詳細は、次のWebサイトにあるKafkaのドキュメンテーションでコンシューマー設定に関するセクション(http://kafka.apache.org/documentation.html#consumerconfigs)をご覧ください。 |
| [Use hierarchical mode] (階層モードを使用) | バイナリ(階層を含む) Avroスキーマを、現在のコンポーネントのスキーマエディターに定義されているフラットスキーマにマップするには、このこのチェックボックスをオンにします。処理するAvroメッセージto be processed is がフラットの場合は、このチェックボックスをオフのままにしておきます。<br><br>オンにする場合は、次のパラメーターを設定する必要があります。<br><br><ul> <li> <p> <span>[Local path to the avro schema] (avroスキーマへのローカルパス)</span>: 処理するAvroデータのスキーマを定義するファイルに移動します。</p> </li> <li> <p> <span>[Mapping]</span>: 現在のコンポーネントのスキーマカラムと処理する階層Avroメッセージに保存されているデータの間にマップを作成します。<span>[Node] (ノード)</span>カラム内に、Avroメッセージから読み取るデータをポイントするJSONパスを入力する必要があります。</p> </li> </ul> |

## 使用方法

| 使用方法のガイダンス | 説明 |
| --- | --- |
| **使用ルール** | このコンポーネントは開始コンポーネントとして使用され、出力リンクを必要とします。<br><br>このコンポーネントは、所属するSpark Streamingコンポーネントのパレットと共に、Spark Streamingジョブを作成している場合にだけ表示されます。<br><br>特に明記していない限り、このドキュメンテーションのシナリオでは、標準ジョブ、つまり従来の Talend Data Integrationジョブだけを扱います。<br><br>Sparkの現在のコンポーネントの実装では、KafkaオフセットはSpark自体によって自動的に管理されます。つまり、ZooKeeperまたはKafkaにコミットされる代わりに、オフセットはSparkチェックポイント内で追跡されます。この実装の詳細は、Sparkのドキュメンテーションで[ダイレクトアプローチ](http://spark.apache.org/docs/latest/streaming-kafka-integration.html)のセクションをご覧ください。 |

**関連概念**

* [スキーマレジストリーから情報を取得](https://help.qlik.com/talend/ja-JP/components/8.0/kafka/retrieving-information-from-schema-registry "このシナリオでは、Spark StreamingジョブでtKafkaInputAvroを使ってスキーマレジストリーからフライト情報を取得する方法を説明します。")