Apache Spark StreamingのtKinesisInputプロパティ

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

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

このコンポーネントのストリーミングバージョンは、Talend Real Time Big Data PlatformおよびTalend Data Fabricで利用できます。

基本設定

[Schema] (スキーマ)と[Edit schema] (スキーマを編集)	スキーマとは行の説明のことです。処理して次のコンポーネントに渡すフィールド(カラム)数を定義します。Sparkジョブを作成する場合、フィールドの命名時は予約語の`line`を避けます。このコンポーネントのスキーマは読み取り専用です。[Edit schema] (スキーマを編集)をクリックすると、スキーマを表示できます。この読み取り専用のpayloadカラムは、処理するKinesisメッセージの本文を保持するために使われます。入力メッセージの本文では、非常に異なるデータ形式が使えます。たとえば、形式がJSONの場合、tKinesisInputに続いてtExtractJSONFieldを使って、この本文から処理するデータを抽出する必要があります。
アクセスキー	AWSアカウントを一意に識別するアクセスキーIDを入力します。アクセスキーとシークレットキーを取得する方法は、http://docs.aws.amazon.com/ses/latest/DeveloperGuide/get-aws-keys.htmlGetting Your AWS Access Keysをご覧ください。
[Secret key] (シークレットキー)	シークレットキーを入力します。シークレットキーは、アクセスキーと組み合わせてセキュリティ認証情報を構成します。パスワードを入力するには、パスワードフィールドの横にある[...]ボタンをクリックし、ポップアップダイアログボックスにパスワードを二重引用符の間に入力し、OKをクリックして設定を保存します。
ストリーム名	tKinesisInputにデータをプルさせるKinesisストリームの名前を入力します。
[Endpoint URL] (エンドポイントURL)	使用するKinesisサービスのエンドポイントを入力します。たとえば、https://kinesis.us-east-1.amazonaws.comです。より有効なKinesisエンドポイントURLは、http://docs.aws.amazon.com/general/latest/gr/rande.html#ak_region (英語のみ)にあります。
[Explicitly set authentication parameters] (認証パラメーターを明示的に設定)	明示的な認証メカニズムを使ってKinesisに接続する場合は、このチェックボックスを選択します。このメカニズムはSpark V1.4以降でのみサポートされています。このセキュリティメカニズムでは、AWSリージョンパラメーターを明示的に設定する必要があるため、使うリージョン値を表示される[Region] (リージョン)フィールドに入力する必要があります。たとえばus-west-2です。使っているSparkバージョンでこのメカニズムがサポートされている場合は、明示的な認証を使ってセキュリティを強化することをお勧めします。このチェックボックスを選択すると、アクセス認証情報がKinesisに直接提供されます。このチェックボックスをオフのままにすると、古い認証メカニズムが使われます。このように、アクセス認証情報はSkinによってKinesis接続のコンテキスト変数として使われます。

詳細設定

[Checkpoint interval] (チェックポイント間隔)	tKinesisInputが読み取りの位置をKinesisストリームに保存する時間間隔(ミリ秒)を入力します。 Kinesisストリームのデータレコードは、パーティション(Kinesisの観点から見たシャード)にグルーピングされ、シーケンス番号でインデックスが付けられます。シーケンス番号は、レコードの位置を一意に識別します。AmazonがKinesisで使う用語の詳細は、http://docs.aws.amazon.com/kinesis/latest/dev/key-concepts.html (英語のみ)をご覧ください。
[Initial position stream] (初期位置のストリーム)	Kinesisチェックポイント情報がない場合にストリームからデータを読み取る開始位置を選択します。 [Start with the oldest data] (最も古いデータで開始): ストリームの最初から24時間以内に開始します。 [Start after the most recent data] (最近のデータの後ろから開始): ストリームの最新データの後ろの位置から始まります。
Storage level (ストレージレベル)	受信したデータをキャッシュする方法を選択します。各レベルの詳細は、https://spark.apache.org/docs/latest/programming-guide.html#rdd-persistence (英語のみ)をご覧ください。

[Checkpoint interval] (チェックポイント間隔)

tKinesisInputが読み取りの位置をKinesisストリームに保存する時間間隔(ミリ秒)を入力します。

Kinesisストリームのデータレコードは、パーティション(Kinesisの観点から見たシャード)にグルーピングされ、シーケンス番号でインデックスが付けられます。シーケンス番号は、レコードの位置を一意に識別します。AmazonがKinesisで使う用語の詳細は、http://docs.aws.amazon.com/kinesis/latest/dev/key-concepts.html (英語のみ)をご覧ください。

[Initial position stream] (初期位置のストリーム)

Kinesisチェックポイント情報がない場合にストリームからデータを読み取る開始位置を選択します。

[Start with the oldest data] (最も古いデータで開始): ストリームの最初から24時間以内に開始します。
[Start after the most recent data] (最近のデータの後ろから開始): ストリームの最新データの後ろの位置から始まります。

Storage level (ストレージレベル)

受信したデータをキャッシュする方法を選択します。各レベルの詳細は、https://spark.apache.org/docs/latest/programming-guide.html#rdd-persistence (英語のみ)をご覧ください。

使用方法

使用ルール	このコンポーネントは、開始コンポーネントとして使用され、出力リンクを必要とします。ランタイムにこのコンポーネントは、ストリームをリッスンし続け、このストリームに新しいメッセージがバッファリングされると読み取ります。このコンポーネントは、所属するSpark Streamingのコンポーネントのパレットと共に、Spark Streamingジョブを作成している場合にだけ表示されます。特に明記していない限り、このドキュメントのシナリオでは、標準ジョブ、つまり従来の Talend データ統合ジョブだけを扱います。
[Spark Connection] (Spark接続)	[Run] (実行)ビューの[Spark configuration] (Spark設定)タブで、ジョブ全体でのSparkクラスターへの接続を定義します。また、ジョブでは、依存jarファイルを実行することを想定しているため、Sparkがこれらのjarファイルにアクセスできるように、これらのファイルの転送先にするファイルシステム内のディレクトリーを指定する必要があります。 Yarnモード(YarnクライアントまたはYarnクラスター): Google Dataprocを使用している場合、[Spark configuration] (Spark設定)タブの[Google Storage staging bucket] (Google Storageステージングバケット)フィールドにバケットを指定します。 HDInsightを使用している場合、[Spark configuration] (Spark設定)タブの[Windows Azure Storage configuration] (Windows Azure Storage設定)エリアでジョブのデプロイメントに使用するブロブを指定します。 Altusを使用する場合は、[Spark configuration] (Spark設定)タブでジョブのデプロイにS3バケットまたはAzure Data Lake Storageを指定します。 Quboleを使用する場合は、ジョブにtS3Configurationを追加し、QuboleでS3システム内に実際のビジネスデータを書き込みます。tS3Configurationを使用しないと、このビジネスデータはQubole HDFSシステムに書き込まれ、クラスターをシャットダウンすると破棄されます。オンプレミスのディストリビューションを使用する場合は、クラスターで使われているファイルシステムに対応する設定コンポーネントを使用します。一般的に、このシステムはHDFSになるため、tHDFSConfigurationを使用します。 [Standalone mode] (スタンドアロンモード): クラスターで使われているファイルシステム(tHDFSConfiguration Apache Spark BatchやtS3Configuration Apache Spark Batchなど)に対応する設定コンポーネントを使用します。ジョブ内に設定コンポーネントがない状態でDatabricksを使用している場合、ビジネスデータはDBFS (Databricks Filesystem)に直接書き込まれます。この接続は、ジョブごとに有効になります。
制限事項	ライセンスの互換性の問題のため、このコンポーネントの使用に必要な1つ以上のJARが提供されていません。この特定のコンポーネントに不足しているJARをインストールするには、Component (コンポーネント)タブビューの[Install] (インストール)ボタンをクリックします。Studioの Integration パースペクティブの[Modules] (モジュール)タブでも、不足しているすべてのJARを簡単に見つけて追加できます。詳細は、外部モジュールのインストールをご覧ください。外部モジュールをインストールする方法は、Talend Help Center (https://help.talend.com (英語のみ))をご覧ください。

使用ルール

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

ランタイムにこのコンポーネントは、ストリームをリッスンし続け、このストリームに新しいメッセージがバッファリングされると読み取ります。

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

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

[Spark Connection] (Spark接続)

[Run] (実行)ビューの[Spark configuration] (Spark設定)タブで、ジョブ全体でのSparkクラスターへの接続を定義します。また、ジョブでは、依存jarファイルを実行することを想定しているため、Sparkがこれらのjarファイルにアクセスできるように、これらのファイルの転送先にするファイルシステム内のディレクトリーを指定する必要があります。

Yarnモード(YarnクライアントまたはYarnクラスター):
- Google Dataprocを使用している場合、[Spark configuration] (Spark設定)タブの[Google Storage staging bucket] (Google Storageステージングバケット)フィールドにバケットを指定します。
- HDInsightを使用している場合、[Spark configuration] (Spark設定)タブの[Windows Azure Storage configuration] (Windows Azure Storage設定)エリアでジョブのデプロイメントに使用するブロブを指定します。
- Altusを使用する場合は、[Spark configuration] (Spark設定)タブでジョブのデプロイにS3バケットまたはAzure Data Lake Storageを指定します。
- Quboleを使用する場合は、ジョブにtS3Configurationを追加し、QuboleでS3システム内に実際のビジネスデータを書き込みます。tS3Configurationを使用しないと、このビジネスデータはQubole HDFSシステムに書き込まれ、クラスターをシャットダウンすると破棄されます。
- オンプレミスのディストリビューションを使用する場合は、クラスターで使われているファイルシステムに対応する設定コンポーネントを使用します。一般的に、このシステムはHDFSになるため、tHDFSConfigurationを使用します。
[Standalone mode] (スタンドアロンモード): クラスターで使われているファイルシステム(tHDFSConfiguration Apache Spark BatchやtS3Configuration Apache Spark Batchなど)に対応する設定コンポーネントを使用します。

ジョブ内に設定コンポーネントがない状態でDatabricksを使用している場合、ビジネスデータはDBFS (Databricks Filesystem)に直接書き込まれます。

この接続は、ジョブごとに有効になります。

制限事項

ライセンスの互換性の問題のため、このコンポーネントの使用に必要な1つ以上のJARが提供されていません。この特定のコンポーネントに不足しているJARをインストールするには、Component (コンポーネント)タブビューの[Install] (インストール)ボタンをクリックします。Studioの Integration パースペクティブの[Modules] (モジュール)タブでも、不足しているすべてのJARを簡単に見つけて追加できます。詳細は、外部モジュールのインストールをご覧ください。外部モジュールをインストールする方法は、Talend Help Center (https://help.talend.com (英語のみ))をご覧ください。

このページは役に立ちましたか?

このページまたはコンテンツにタイポ、ステップの省略、技術的エラーなどの問題が見つかった場合はお知らせください。

こちらにフィードバックをお寄せください