314.6. DataFrame 작업

RDDs Spark 구성 요소로 작업하는 대신 DataFrames에서도 작업할 수 있습니다.

DataFrame 작업을 호출하려면 다음 URI를 사용합니다.

스파크 RDD 프로듀서

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

 Where `dataFrame` option refers to the name of an DataFrame instance
(`instances of org.apache.spark.sql.Dataset and org.apache.spark.sql.Row`) from a Camel registry,
while `dataFrameCallback` refers to the implementation
of `org.apache.camel.component.spark.DataFrameCallback` interface (also
from a registry). DataFrame callback provides a single method used to
apply incoming messages against the given DataFrame. Results of callback
computations are saved as a body to an exchange.

스파크 RDD 콜백

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

다음 스니펫에서는 메시지를 작업에 입력으로 보내고 결과를 반환하는 방법을 보여줍니다.

스파크 작업 호출

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

Springans로 등록된 코드 조각에 대한 DataFrame 콜백은 다음과 같습니다.

스파크 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 정의는 다음과 같습니다.

스파크 RDD 정의

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