314.5. DataFrame 작업

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

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

Spark RDD 생산자

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

 여기서 dataFrame 옵션은 Camel 레지스트리에서 DataFrame인스턴스( org.apache.spark.sql.Dataset 및 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 빈으로 등록된 위의 조각에 대한 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;
}