309.5. DataFrame ジョブ

RDD を使用する代わりに、Spark コンポーネントは DataFrame を使用することもできます。

DataFrame ジョブを呼び出すには、次の URI を使用します。

Spark RDD producer

spark:dataframe?dataFrame=#testDataFrame&dataFrameCallback=#transformation

spark:dataframe?dataFrame=#testDataFrame&dataFrameCallback=#transformation

Copy to Clipboard

Toggle word wrap

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);
}

public interface DataFrameCallback<T> {
    T onDataFrame(Dataset<Row> dataFrame, Object... payloads);
}

Copy to Clipboard

Toggle word wrap

次のスニペットは、ジョブへの入力としてメッセージを送信し、結果を返す方法を示しています。

Spark ジョブの呼び出し

String model = "Micra";
long linesCount = producerTemplate.requestBody("spark:dataFrame?dataFrame=#cars&dataFrameCallback=#findCarWithModel", model, long.class);

String model = "Micra";
long linesCount = producerTemplate.requestBody("spark:dataFrame?dataFrame=#cars&dataFrameCallback=#findCarWithModel", model, long.class);

Copy to Clipboard

Toggle word wrap

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();
        }
    };
}

@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();
        }
    };
}

Copy to Clipboard

Toggle word wrap

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;
}

@Bean
Dataset<Row> cars(HiveContext hiveContext) {
    Dataset<Row> jsonCars = hiveContext.read().json("/var/data/cars.json");
    jsonCars.registerTempTable("cars");
    return jsonCars;
}

Copy to Clipboard

Toggle word wrap

トップに戻る

309.5. DataFrame ジョブ

詳細情報

試用、購入および販売

コミュニティー

Red Hat ドキュメントについて

多様性を受け入れるオープンソースの強化

会社概要

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links