在本地模式下運行一個簡單的級聯程序

Question

我很努力讓這個簡單的級聯程序運行。由於某種原因，它什麼都不做。最不希望它打印記錄。任何幫助，將不勝感激。

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

您沒有執行流程。

創建流程後，請致電complete()（阻止）或start()執行它。調用stop()是而不是要去執行流程。

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