313.5. DataFrame ジョブ
RDD を使用する代わりに、Spark コンポーネントは DataFrame を使用することもできます。
DataFrame ジョブを呼び出すには、次の URI を使用します。
Spark RDD producer
spark:dataframe?dataFrame=#testDataFrame&dataFrameCallback=#transformation
dataFrame オプションは、Camel レジストリーの DataFrame インスタンス (instances of org.apache.spark.sql.Dataset and org.apache.spark.sql.Row) の名前を参照し、dataFrameCallback は org.apache.camel.component.spark.DataFrameCallback インターフェイスの実装を参照します (レジストリーからも)。DataFrame コールバックは、指定された DataFrame に対して入力メッセージを適用するために使用される単一のメソッドを提供します。コールバック計算の結果は、ボディーとしてエクスチェンジに保存されます。
Spark RDD コールバック
public interface DataFrameCallback<T> {
T onDataFrame(Dataset<Row> dataFrame, Object... payloads);
}次のスニペットは、ジョブへの入力としてメッセージを送信し、結果を返す方法を示しています。
Spark ジョブの呼び出し
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(Dataset<Row> dataFrame, Object... payloads) {
String model = (String) payloads[0];
return dataFrame.where(dataFrame.col("model").eqNullSafe(model)).count();
}
};
}Spring の DataFrame 定義は次のようになります。
Spark RDD 定義
@Bean
Dataset<Row> cars(HiveContext hiveContext) {
Dataset<Row> jsonCars = hiveContext.read().json("/var/data/cars.json");
jsonCars.registerTempTable("cars");
return jsonCars;
}