313.5. DataECDHE 작업

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

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

스파크 RDD 프로듀서

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

 여기서 data DestinationRule 옵션은 Camel 레지스트리의 org.apache.spark.sql.Dataset 및 org.apache.spark.sql.Row )의 데이터 DestinationRule 인스턴스이름을 참조하는 반면, data ECDHEECDHEback은 org.apache.camel.component.Data ECDHEback 인터페이스 (또는 레지스트리에서도)의 구현을 나타냅니다. DataECDHE 콜백은 지정된 DataECDHE에 대해 수신되는 메시지를 적용하는 데 사용되는 단일 메서드를 제공합니다. 콜백 계산의 결과는 교환에 본문으로 저장됩니다.

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

SpringBean으로 등록된 위의 스니펫에 대한 DataECDHE 콜백은 다음과 같습니다.

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

Spark RDD 정의

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