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