Hi team,
I am trying to process some data using Operators.
@SuppressWarnings("unchecked")
@Override
public void populateDAG(DAG dag, Configuration conf) {
System.setProperty("viewmode", "production");
CouchBasePOJOInputOperator inputOperator =
dag.addOperator("inputOperator", CouchBasePOJOInputOperator.class);
inputOperator.setStore(new CouchBaseStore());
MedOperator med = dag.addOperator("median", MedOperator.class);
MeanOperator mean=dag.addOperator("mean", MeanOperator.class);
StandardDeviationOperator sdo=dag.addOperator("sdo",
StandardDeviationOperator.class);
ConsoleOutputOperator cons = dag.addOperator("cons", new
ConsoleOutputOperator());
ConsoleOutputOperator cons1 = dag.addOperator("cons1", new
ConsoleOutputOperator());
ConsoleOutputOperator cons2 = dag.addOperator("cons2", new
ConsoleOutputOperator());
dag.addStream("inputFormatter", inputOperator.outputPort,
med.data,mean.meandata,sdo.meandata);
dag.addStream("cons", med.median,
cons.input).setLocality(Locality.THREAD_LOCAL);
dag.addStream("cons1", mean.mean,
cons1.input).setLocality(Locality.THREAD_LOCAL);
dag.addStream("cons2", sdo.deviation,
cons2.input).setLocality(Locality.THREAD_LOCAL);
}
There is no error in the code but when I launch this application in Data
Torrent, the status of operators remains pending instead of running.
Regards,
Hitesh Goyal
Simpli5d Technologies
Cont No.: 9599803307