import cascading.operation.Identity; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void complete() {
  List<TupleEntry> actual = new FunctionCallStub.Builder(FIELDS)
      .addTuple("a")
      .addTuple("b")
      .build()
      .complete(mock(FlowProcess.class), new Identity())
      .result()
      .asTupleEntryList();

  assertThat(actual.size(), is(2));
  assertThat(actual.get(0), tupleEntry(FIELDS, "a"));
  assertThat(actual.get(1), tupleEntry(FIELDS, "b"));
}
 

import cascading.operation.Identity; //导入依赖的package包/类
@Test
public void complexPipe() throws Exception {
    Pipe pipe = PipeBuilder.start()
            .each().select("data").filterOut().nullValues()
            .each().insertField("processingDate").withValue(System.currentTimeMillis())
            .renameField("domains").to("oldDomains")
            .each().select("url").applyFunction(new ExpressionFunction(new Fields("isRelativeUrl"), "url.startsWith(\"http\")", String.class))
            .compressOutput().forThisStepOnly().withGzipCodec().withType("BLOCK").ofTheMappers()
            .discard("data")
            .groupBy().withSortOnFields("url").onFields("domain")
            .count("domains")
            .pipe();
    
    // count
    assertThat(pipe).isInstanceOf(Every.class);
    assertThat(((Every)pipe).getOperation()).isInstanceOf(Count.class);
    
    // groupBy
    pipe = pipe.getPrevious()[0];
    assertThat(pipe).isInstanceOf(GroupBy.class);
    
    // discard
    pipe = pipe.getPrevious()[0];
    assertThat(pipe).isInstanceOf(Discard.class);
    pipe = pipe.getPrevious()[0];
    assertThat(pipe).isInstanceOf(Each.class);
    assertThat(((Each)pipe).getOperation()).isInstanceOf(NoOp.class);
    
    // expression
    pipe = pipe.getPrevious()[0];
    assertThat(pipe).isInstanceOf(Each.class);
    assertThat(((Each)pipe).getOperation()).isInstanceOf(ExpressionFunction.class);
    
    // rename
    pipe = pipe.getPrevious()[0];
    assertThat(pipe).isInstanceOf(Rename.class);
    pipe = pipe.getPrevious()[0];
    assertThat(pipe).isInstanceOf(Each.class);
    assertThat(((Each)pipe).getOperation()).isInstanceOf(Identity.class);
    
    // insert
    pipe = pipe.getPrevious()[0];
    assertThat(pipe).isInstanceOf(Each.class);
    assertThat(((Each)pipe).getOperation()).isInstanceOf(Insert.class);
    
    // filter null
    pipe = pipe.getPrevious()[0];
    assertThat(pipe).isInstanceOf(Each.class);
    assertThat(((Each)pipe).getOperation()).isInstanceOf(FilterNull.class);
    
    // initial pipe
    pipe = pipe.getPrevious()[0];
    assertThat(pipe.getClass()).isEqualTo(Pipe.class);
}