Führen Sie ein einfaches Cascading-Programm im lokalen Modus

Question

Ich habe Schwierigkeiten, dieses einfache kaskadierende Programm zu starten. Aus irgendeinem Grund tut es nichts. Zumindest würde ich erwarten, dass es die Aufzeichnungen druckt. Jede Hilfe wäre willkommen.

package com.myLearning.cascading; 

import cascading.flow.Flow; 
import cascading.flow.FlowDef; 
import cascading.flow.local.LocalFlowConnector; 
import cascading.operation.Debug; 
import cascading.operation.expression.ExpressionFilter; 
import cascading.pipe.Each; 
import cascading.pipe.Pipe; 
import cascading.scheme.Scheme; 
import cascading.scheme.local.TextDelimited; 
import cascading.tap.SinkMode; 
import cascading.tap.Tap; 
import cascading.tap.local.FileTap; 
import cascading.tuple.Fields; 

public class operations_example 
{ 
    public static void main(String[] args) 
    { 
    Scheme sourceScheme = new TextDelimited(new Fields("username", "age"), true, ","); 
    String sourcePath = "C:/Users/Desktop/cascading/data/names.txt"; 
    Tap sourceTap = new FileTap(sourceScheme, sourcePath); 

    Scheme targetScheme = new TextDelimited(new Fields("username", "age"), true, ","); 
    String targetPath = "C:/Users/Desktop/cascading/data/output2.txt"; 
    Tap targetTap = new FileTap(targetScheme, targetPath, SinkMode.REPLACE); 

    Pipe dataPipe = new Pipe("data"); 
    dataPipe = new Each(dataPipe, new Debug()); 
    ExpressionFilter filter = new ExpressionFilter("age >= 30", Integer.TYPE); 

    dataPipe = new Each(dataPipe,new Fields("username","age"), filter); 

    FlowDef flowdef = FlowDef.flowDef(). 
      addSource(dataPipe, sourceTap). 
      addTailSink(dataPipe, targetTap); 

    Flow flow = new LocalFlowConnector().connect(flowdef); 
    flow.stop();  
    } 
} 

Answer 1

Sie haben den Fluss nicht ausgeführt.

Nachdem der Fluss erstellt wurde, rufen Sie entweder complete() (Blockierung) oder start() auf, um ihn auszuführen. Rufen Sie stop() ist nicht gehen, um den Fluss auszuführen.

http://docs.cascading.org/impatient/impatient1.html http://docs.cascading.org/cascading/1.2/javadoc/cascading/flow/Flow.html#complete()