292.4. DataFrame ジョブ
RDDs Spark コンポーネントを使用する代わりに、DataFrame と連携することもできます。
DataFrame ジョブを呼び出すには、以下の URI を使用します。
Spark RDD プロデューサー
spark:dataframe?dataFrame=#testDataFrame&dataFrameCallback=#transformation
dataFrame オプションは、Camel レジストリーからの DataFrame インスタンスの名前(org.apache.spark.sql.DataFrame のインスタンス)を参照しますが、 dataFrameCallback は org.apache.camel.component.spark.DataFrameCallback インターフェースの実装を指します(レジストリーとも呼ばれます)。DataFrame コールバックは、指定の DataFrame に受信メッセージを適用する単一の方法を提供します。コールバック計算の結果は、エクスチェンジのボディーとして保存されます。
Spark RDD コールバック
public interface DataFrameCallback<T> {
T onDataFrame(DataFrame dataFrame, Object... payloads);
}
以下のスニペットは、メッセージをジョブへの入力として送信し、結果を返す方法を示しています。
スパークジョブの呼び出し
String model = "Micra";
long linesCount = producerTemplate.requestBody("spark:dataFrame?dataFrame=#cars&dataFrameCallback=#findCarWithModel", model, long.class);
Spring Bean として登録された上記のスニペットの DataFrame コールバックは、以下のようになります。
Spark RDD コールバック
@Bean
RddCallback<Long> findCarWithModel() {
return new DataFrameCallback<Long>() {
@Override
public Long onDataFrame(DataFrame dataFrame, Object... payloads) {
String model = (String) payloads[0];
return dataFrame.where(dataFrame.col("model").eqNullSafe(model)).count();
}
};
}
Spring の DataFrame 定義は以下のようになります。
Spark RDD の定義
@Bean
DataFrame cars(HiveContext hiveContext) {
DataFrame jsonCars = hiveContext.read().json("/var/data/cars.json");
jsonCars.registerTempTable("cars");
return jsonCars;
}