C Sparkナレッジ・モジュール

この付録では、Sparkナレッジ・モジュールについて説明します。

この章の内容は次のとおりです。

• LKM File to Spark

• LKM Spark to File

• LKM Hive to Spark

• LKM Spark to Hive

• LKM HDFS to Spark

• LKM Spark to HDFS

• LKM Kafka to Spark

• LKM Spark to Kafka

• LKM SQL to Spark

• LKM Spark to SQL

• RKM Cassandra

• XKM Spark Aggregate

• XKM Spark Distinct

• XKM Spark Expression

• XKM Spark Filter

• XKM Spark Flatten

• XKM Spark Input SignatureおよびOutput Signature

• XKM Spark Join

• XKM Spark Lookup

• XKM Spark Pivot

• XKM Spark Set

• XKM Spark Sort

• XKM Spark Split

• XKM Spark Table Function

• IKM Spark Table Function

• XKM Spark Unpivot

C.1 LKM File to Spark

このKMはデータをファイルからSpark Python変数にロードし、実行ユニット、ソース・テクノロジであるファイル、ターゲット・テクノロジであるSpark Pythonの間のAPで定義できます。

次の表は、LKM File to Sparkのオプションについて説明します。

---

表C-1 LKM File to Spark

オプション	説明
ストレージ関数	データをロード/保存するのに使用されるストレージ関数。
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。
InputFormatClass	Hadoop InputFormatのクラス名。 例: org.apache.hadoop.mapreduce.lib.input.TextInputFormat。
KeyClass	キーWritableクラスの完全修飾クラス名。 例: org.apache.hadoop.io.Text。
ValueClass	値Writableクラスの完全修飾クラス名。 例: org.apache.hadoop.io.LongWritable。
KeyConverter	キー・コンバータ・クラスの完全修飾クラス名。
ValueConverter	値コンバータ・クラスの完全修飾クラス名。
ジョブ構成	Hadoop構成。 例: {'hbase.zookeeper.quorum': 'HOST', 'hbase.mapreduce.inputtable': 'TAB'}

---

このLKMは、StreamingContext.textFileStream()メソッドを使用して、ファイル・コンテキストをデータ・ストリームとして転送します。Sparkアプリケーションの実行中、ディレクトリはモニターされます。他の場所からこのディレクトリにコピーされたファイルはすべて削除されます。

---

表C-2 ストリーミング用のLKM File to Spark

オプション	説明
STREAMING_MODE	このオプションは、マッピングをストリーミング・モードで実行するかどうかを示します。 デフォルトはFALSEです。
ストレージ関数	STREAMING_MODEをtrueに設定すると、ロード関数はtextFileStreamに自動的に変更されます。 デフォルトはtextFileです。
ソース・データ・ストア	ソース・データ・ストアはディレクトリであり、フィールド・セパレータを定義する必要があります。

---

C.2 LKM Spark to File

このKMはデータをSpark Python変数からファイルに保存し、実行ユニット、ソース・テクノロジであるSpark Python、ターゲット・テクノロジであるファイルの間のAPで定義できます。

次の表は、LKM Spark to Fileのオプションについて説明します。

---

表C-3 LKM Spark to File

オプション	説明
ストレージ関数	データをロード/保存するのに使用されるストレージ関数。
InputFormatClass	Hadoop InputFormatのクラス名。 例: org.apache.hadoop.mapreduce.lib.input.TextInputFormat。
KeyClass	キーWritableクラスの完全修飾クラス名。 例: org.apache.hadoop.io.Text。
ValueClass	値Writableクラスの完全修飾クラス名。 例: org.apache.hadoop.io.LongWritable。
KeyConverter	キー・コンバータ・クラスの完全修飾クラス名。
ValueConverter	値コンバータ・クラスの完全修飾クラス名。
ジョブ構成	Hadoop構成。 例: {'hbase.zookeeper.quorum': 'HOST', 'hbase.mapreduce.inputtable': 'TAB'}

---

---

表C-4 ストリーミング用のLKM Spark to File

オプション	説明
ストレージ関数	STREAMING_MODEをtrueに設定すると、ロード関数はtextFileStreamに自動的に変更されます。 デフォルトはtextFileです。

---

C.3 LKM Hive to Spark

このKMはデータをHive表からSpark Python変数にロードし、実行ユニット、ソース・テクノロジであるHive、ターゲット・テクノロジであるSpark Pythonの間のAPで定義できます。

C.4 LKM Spark to Hive

このKMはデータをSpark Python変数からHive表に保存し、実行ユニット、ソース・テクノロジであるSpark Python、ターゲット・テクノロジであるHiveの間のAPで定義できます。

次の表は、LKM Spark to Hiveのオプションについて説明します。

---

表C-5 LKM Spark to Hive

オプション	説明
CREATE_TARGET_TABLE	ターゲット表を作成します。
OVERWRITE_TARGET_TABLE	ターゲット表を上書きします。

---

C.5 LKM HDFS to Spark

このKMは、データをHDFSファイルからSparkにロードします。

---

表C-6 LKM HDFS to Spark

オプション	説明
ストレージ関数	ストレージ関数は、データのロードまたは保存に使用されます。
streamingContext	ストリーミング・コンテキストの名前。
InputFormatClass	Hadoop InputFormatのクラス名。 例: org.apache.hadoop.mapreduce.lib.input.TextInputFormat。
KeyClass	キーWritableクラスの完全修飾クラス名。 例: org.apache.hadoop.io.Text。
ValueClass	値Writableクラスの完全修飾クラス名。 例: org.apache.hadoop.io.LongWritable。
KeyConverter	キー・コンバータ・クラスの完全修飾クラス名。
ValueConverter	値コンバータ・クラスの完全修飾クラス名。
ジョブ構成	Hadoop構成。 例: {'hbase.zookeeper.quorum': 'HOST', 'hbase.mapreduce.inputtable': 'TAB'}
Sparkマッピング・ファイルの削除	マッピングの最後に一時オブジェクトを削除します。
キャッシュ	計算後に、操作全体でRDDをキャッシュします。
ストレージ・レベル	ストレージ・レベルは、データのキャッシュに使用されます。
再パーティション	このコンポーネントの変換後に、RDDを再パーティション化します。
並列度のレベル	パーティションの数。
パーティションのソート	RDDを再パーティション化する際に、パーティションをキー関数でソートします。
パーティションのソート順序	パーティションをソートする順序
パーティション・キー関数	パーティションのキーを定義します。
パーティション関数	カスタマイズされたパーティション関数。

---

C.6 LKM Spark to HDFS

このKMは、データをSparkからHDFSファイルにロードします。

---

表C-7 LKM Spark to HDFS

オプション	説明
ストレージ関数	ストレージ関数は、データのロードまたは保存に使用されます。
OutputFormatClass	Hadoop入力書式のクラス名。
KeyClass	キーWritableクラスの完全修飾クラス名。 例: org.apache.hadoop.io.Text。
ValueClass	値Writableクラスの完全修飾クラス名。 例: org.apache.hadoop.io.LongWritable。
KeyConverter	キー・コンバータ・クラスの完全修飾クラス名。
ValueConverter	値コンバータ・クラスの完全修飾クラス名。
ジョブ構成	Hadoop構成。 例: {'hbase.zookeeper.quorum': 'HOST', 'hbase.mapreduce.inputtable': 'TAB'}
DELETE_TEMPORARY_OBJECTS	マッピングの最後に一時オブジェクトを削除します。
Sparkマッピング・ファイルの削除	マッピングの最後に一時オブジェクトを削除します。
キャッシュ	計算後に、操作全体でRDDをキャッシュします。
ストレージ・レベル	ストレージ・レベルは、データのキャッシュに使用されます。
再パーティション	このコンポーネントの変換後に、RDDを再パーティション化します。
並列度のレベル	パーティションの数。
パーティションのソート	RDDを再パーティション化する際に、パーティションをキー関数でソートします。
パーティションのソート順序	パーティションをソートする順序
パーティション・キー関数	パーティションのキーを定義します。
パーティション関数	カスタマイズされたパーティション関数。

---

C.7 LKM Kafka to Spark

このKMは、KafkaソースおよびSparkターゲットを使用してデータをロードし、Spark実行ユニットに存在してKafkaアップストリーム・ノードを保持するAPノードで定義できます。

---

表C-8 ストリーミング用のLKM Kafka to Spark

オプション	説明
ストレージ関数	STREAMING_MODEをtrueに設定すると、ロード関数はtextFileStreamに自動的に変更されます。 デフォルトはcreateStreamです。
キー・デコーダ	メッセージ・キーをデコードします。 デフォルトは空です。
値デコーダ	メッセージ値をデコードします。 デフォルトは空です。
グループID	KafkaUtils.createStreamのコールに使用される受信者グループIDパラメータ。 注意: 受信者のグループ(すべての受信者のグループIDが同じ)では、すべてのメッセージを単一の受信者のみが受信します。
Kakfaパラメータ	KafkaUitls.createStreamのコールに使用されるパラメータ。
storageLevel	ストレージ・レベルは、受信されたオブジェクトの格納に使用されます。 デフォルトはStorageLevel.MEMORY_AND_DISK_2です。
パーティションの数	各スレッドがKafkaからデータを取得するパーティションの数
開始オフセット	createDirectStream関数とともに使用されるパラメータ。

---

C.8 LKM Spark to Kafka

LKM Spark to Kafkaは、ストリーミングとバッチの両方のモードで機能し、実行ユニット間に存在してKafkaダウンストリーム・ノードを保持するAPで定義できます。

---

表C-9 LKM Spark to Kafka

オプション	説明
value.serializer	org.apache.kafka.common.serialization.StringSerializer

---

C.9 LKM SQL to Spark

このKMは、データをCassandraからSparkにロードするよう設計されていますが、他のJDBCソースとでも機能します。SQLソースおよびSparkターゲットを保持するAPノードで定義できます。

---

表C-10 LKM SQL to Spark

オプション	説明
PARTITION_COLUMN	パーティション化に使用される列
LOWER_BOUND	パーティション列の下限。
UPPER_BOUND	パーティション列の上限。
NUMBER_PARTITIONS	パーティションの数。
PREDICATES	述語のリスト。

---

C.10 LKM Spark to SQL

このKMは、データをSparkからCassandra表にロードし、SparkソースおよびSQLターゲットを保持するAPノードで定義できます。他のJDBCターゲットとでも機能します。

---

表C-11 LKM Spark to SQL

オプション	説明
CREATE_TARG_TABLE	ターゲット表を作成します。

---

C.11 RKM Cassandra

RKM Cassandraでは、次のメタデータ要素がリバースされます。

• データストアとしてのCassandra表。

  「リバース・エンジニアリング」タブの「マスク」フィールドで、リバースエンジニアリングされるオブジェクトが名前に基づいてフィルタ処理されます。「マスク」フィールドは空にできず、少なくともパーセント記号(%)を含める必要があります。

• 属性としてのCassandra列およびそのデータ型。

C.12 XKM Spark Aggregate

SUMやGROUP BYなどを使用して行を集約します。

次の表は、XKM Spark Aggregateのオプションについて説明します。

---

表C-12 XKM Spark Aggregate

オプション	説明
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。
NUMBER_OF_TASKS	タスク番号。

---

---

表C-13 ストリーミング用のXKM Spark Aggregate

オプション	説明
WINDOW_AGGREGATION	ウィンドウ集計を有効にします。
WINDOW_LENGTH	バッチ間隔の数。
SLIDING_INTERVAL	ウィンドウ操作が実行される間隔。
STATEFUL_AGGREGATION	ステートフル集計を有効にします。
STATE_RETENTION_PERIOD	Spark状態オブジェクトでキーまたは値の集計を保持する時間(秒)。
FORWARD_ONLY_UPDATED_ROWS	ダウンストリーム・コンポーネントに転送される変更された集計値。

---

C.13 XKM Spark Distinct

データの重複を排除し、機能は既存のバッチ処理と同一です。

C.14 XKM Spark Expression

単一のマッピングにおいて再使用される式を定義します。

C.15 XKM Spark Filter

フィルタ条件に基づいてデータのサブセットを生成します。

次の表は、XKM Spark Filterのオプションについて説明します。

---

表C-14 XKM Spark Filter

オプション	説明
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。

---

C.16 XKM Spark Input SignatureおよびOutput Signature

再使用可能マッピングのコード生成がサポートされています。

C.17 XKM Spark Join

結合条件に基づいて複数の入力ソースを結合します。

次の表は、XKM Spark Joinのオプションについて説明します。

---

表C-15 XKM Spark Join

オプション	説明
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。
NUMBER_OF_TASKS	タスク番号。

---

C.18 XKM Spark Lookup

派生データ・ソースのデータをルックアップします。

次の表は、XKM Spark Lookupのオプションについて説明します。

---

表C-16 XKM Spark Lookup

オプション	説明
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。
NUMBER_OF_TASKS	タスク番号。
MAP_SIDE	KMでマップ側ルックアップを実行するのかリデュース側ルックアップを実行するのかを定義し、ルックアップのパフォーマンスにきわめて大きな影響を及ぼします。
KEY_BASED_LOOKUP	ルックアップ・キーに対応するデータのみが取得されます。

---

---

表C-17 ストリーミング用のXKM Spark Lookup

オプション	説明
MAP_SIDE	MAP_SIDE=true: メモリーに入る小さいルックアップ・データ・セットに適しています。この設定により、ルックアップ・データがすべてのSparkタスクにブロードキャストされてパフォーマンスが向上します。
KEY_BASED_LOOKUP	どの着信ルックアップ・キーについても、Sparkキャッシュがチェックされます。 ルックアップ・レコードが存在し、失効していない場合、ルックアップ・データはキャッシュから提供されます。 ルックアップ・レコードが存在しないか、失効している場合、データはSQLソースから再ロードされます。
CACHE_RELOAD	このオプションは、ルックアップ・ソース・データがいつロードおよびリフレッシュされるかを定義します。対応する値は次のとおりです。 NO_RELOAD: ルックアップ・ソース・データは、Sparkアプリケーションの起動時に1回ロードされます。 RELOAD_EVERY_BATCH: ルックアップ・ソース・データは、新しいSparkバッチのたびに再ロードされます。 RELOAD_BASE_ON_TIME: ルックアップ・ソース・データは、Sparkアプリケーションの起動時にロードされ、KMオプションのCacheReloadIntervalで指定された時間間隔後にリフレッシュされます。
CACHE_RELOAD_INTERVAL	データがSparkキャッシュで保持される時間を定義します。この時間を経過すると、失効したデータまたはレコードはキャッシュから削除されます。

---

C.19 XKM Spark Pivot

別々の行のデータを取得して集計し、列に変換します。

次の表は、XKM Spark Pivotのオプションについて説明します。

---

表C-18 XKM Spark Pivot

オプション	説明
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。

---

注意:

XKM Spark Pivotでは、ストリーミングがサポートされていません。

C.20 XKM Spark Set

UNION、MINUSまたはその他の集合演算を実行します。

C.21 XKM Spark Sort

式を使用してデータをソートします。

次の表は、XKM Spark Sortのオプションについて説明します。

---

表C-19 XKM Spark Sort

オプション	説明
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。
NUMBER_OF_TASKS	タスク番号。

---

C.22 XKM Spark Split

データを複数の条件により複数のパスに分割します。

次の表は、XKM Spark Splitのオプションについて説明します。

---

表C-20 XKM Spark Split

オプション	説明
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。

---

C.23 XKM Spark Table Function

Spark表関数アクセス。

次の表は、XKM Spark Table Functionのオプションについて説明します。

---

表C-21 XKM Spark Table Function

オプション	説明
SPARK_SCRIPT_FILE	ユーザーがsparkスクリプト・ファイルのパスを指定します。
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。

---

C.24 IKM Spark Table Function

ターゲットとしてのSpark表関数。

次の表は、IKM Spark Table Functionのオプションについて説明します。

---

表C-22 IKM Spark Table Function

オプション	説明
SPARK_SCRIPT_FILE	ユーザーがsparkスクリプト・ファイルのパスを指定します。
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。

---

C.25 XKM Spark Unpivot

複数の属性の単一行を複数の行に効率的に変換します。

次の表は、XKM Spark Pivotのオプションについて説明します。

---

表C-23 XKM Spark Unpivot

オプション	説明
CACHE_DATA	デフォルト・ストレージ・レベルを使用してデータを永続化します。

---

注意:

XKM Spark Unpivotでは、ストリーミングがサポートされていません。