import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openTrapForWrite(Tap trap) throws IOException {

	if (trap instanceof Hfs) {

		JobConf jobConf = new JobConf(this.getConfigCopy());

		int stepNum = jobConf.getInt( "cascading.flow.step.num", 0 );
		int nodeNum = jobConf.getInt( "cascading.flow.node.num", 0 );

		String partname = String.format( "-%05d-%05d-%05d", stepNum, nodeNum, this.getCurrentSliceNum() );
		jobConf.set( "cascading.tapcollector.partname", "%s%spart" + partname );

		String value = String.format( "attempt_%012d_0000_m_%06d_0", (int) Math.rint( System.currentTimeMillis() ), this.getCurrentSliceNum() );
		jobConf.set( "mapred.task.id", value );
		jobConf.set( "mapreduce.task.id", value );

		return trap.openForWrite( new FlinkFlowProcess( jobConf ), null);
	}
	else {
		throw new UnsupportedOperationException("Only Hfs taps are supported as traps");
	}
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
private void writeToHadoopPartitionTap(Tap<?, ?, ?> tap) throws IOException {
  @SuppressWarnings("unchecked")
  BasePartitionTap<JobConf, ?, ?> hadoopTap = (BasePartitionTap<JobConf, ?, ?>) tap;
  JobConf conf = new JobConf();

  // Avoids deletion of results when using a partition tap (close() will delete the _temporary before the copy has
  // been done if not in a flow)
  HadoopUtil.setIsInflow(conf);

  HadoopFlowProcess flowProcess = new HadoopFlowProcess(conf);
  hadoopTap.sinkConfInit(flowProcess, conf);
  TupleEntryCollector collector = hadoopTap.openForWrite(flowProcess);
  for (TupleEntry tuple : data.asTupleEntryList()) {
    collector.add(tuple);
  }
  collector.close();

  // We need to clean up the '_temporary' folder
  BasePartitionTap<JobConf, ?, ?> partitionTap = hadoopTap;
  @SuppressWarnings("unchecked")
  String basePath = partitionTap.getParent().getFullIdentifier(flowProcess);
  deleteTemporaryPath(new Path(basePath), FileSystem.get(conf));
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
private void writeToLocalTap(Tap<?, ?, ?> tap) throws IOException {
  @SuppressWarnings("unchecked")
  Tap<Properties, ?, ?> localTap = (Tap<Properties, ?, ?>) tap;
  Properties conf = new Properties();
  LocalFlowProcess flowProcess = new LocalFlowProcess(conf);

  flowProcess.setStepStats(new LocalStepStats(new NullFlowStep(), NullClientState.INSTANCE));

  localTap.sinkConfInit(flowProcess, conf);
  TupleEntryCollector collector = localTap.openForWrite(flowProcess);
  for (TupleEntry tuple : data.asTupleEntryList()) {
    collector.add(tuple);
  }
  collector.close();
  localTap.commitResource(conf);
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
private TupleEntryCollector getCollector(String path) {
  TupleEntryCollector collector = collectors.get(path);

  if (collector != null)
    return collector;

  try {

    collector = createTupleEntrySchemeCollector(flowProcess,
        parent, path);

    flowProcess.increment(Counters.Paths_Opened, 1);
  } catch (IOException exception) {
    throw new TapException("unable to open template path: " + path,
        exception);
  }

  if (collectors.size() > openTapsThreshold)
    purgeCollectors();

  collectors.put(path, collector);

  return collector;
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Test
public void exerciseScheme() throws IOException {
  TupleEntryCollector collector = tap.openForWrite(flowProcess);
  for (Tuple tuple : tuples) {
    collector.add(tuple);
  }
  collector.close();
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
public TupleEntryCollector openForWrite(FlowProcess<Config> flowProcess, Object output) throws IOException {
    return new RedisSchemeCollector(flowProcess,
                                    (RedisBaseScheme) getScheme(),
                                    hostname,
                                    port,
                                    db);
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
private void writeToHadoopTap(Tap<?, ?, ?> tap) throws IOException {
  @SuppressWarnings("unchecked")
  Tap<JobConf, ?, ?> hadoopTap = (Tap<JobConf, ?, ?>) tap;
  JobConf conf = new JobConf();

  HadoopFlowProcess flowProcess = new HadoopFlowProcess(conf);
  hadoopTap.sinkConfInit(flowProcess, conf);
  TupleEntryCollector collector = hadoopTap.openForWrite(flowProcess);
  for (TupleEntry tuple : data.asTupleEntryList()) {
    collector.add(tuple);
  }
  collector.close();
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Test
public void asTupleEntryList() throws IOException {
  Bucket sink = new Bucket(FIELDS, pipe, flow);
  TupleEntryCollector collector = sink.openForWrite(null, null);
  collector.add(TUPLE_1);
  collector.add(TUPLE_2);
  List<TupleEntry> tupleEntryList = sink.result().asTupleEntryList();
  assertThat(tupleEntryList.size(), is(2));
  assertThat(tupleEntryList.get(0).getFields(), is(FIELDS));
  assertThat(tupleEntryList.get(0).getTuple(), is(TUPLE_1));
  assertThat(tupleEntryList.get(1).getFields(), is(FIELDS));
  assertThat(tupleEntryList.get(1).getTuple(), is(TUPLE_2));
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Test
public void asTupleList() throws IOException {
  Bucket sink = new Bucket(FIELDS, pipe, flow);
  TupleEntryCollector collector = sink.openForWrite(null, null);
  collector.add(TUPLE_1);
  collector.add(TUPLE_2);
  List<Tuple> tupleList = sink.result().asTupleList();
  assertThat(tupleList.size(), is(2));
  assertThat(tupleList.get(0), is(TUPLE_1));
  assertThat(tupleList.get(1), is(TUPLE_2));
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public void close() {
  super.close();

  try {
    for (TupleEntryCollector collector : collectors.values())
      closeCollector(collector);
  } finally {
    collectors.clear();
  }
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
private void closeCollector(TupleEntryCollector collector) {
  if (collector == null)
    return;

  try {
    collector.close();

    flowProcess.increment(Counters.Paths_Closed, 1);
  } catch (Exception exception) {
    // do nothing
  }
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openForWrite( FlowProcess<JobConf> flowProcess, OutputCollector output ) throws IOException {
    if( !isSink() )
        throw new TapException( "this tap may not be used as a sink, no TableDesc defined" );

    LOG.info("Creating JDBCTapCollector output instance");
    JDBCTapCollector jdbcCollector = new JDBCTapCollector( flowProcess, this );

    jdbcCollector.prepare();

    return jdbcCollector;
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openForWrite(FlowProcess<JobConf> jobConfFlowProcess, OutputCollector outputCollector)
		throws IOException {
	HBaseTapCollector hBaseCollector = new HBaseTapCollector(jobConfFlowProcess, this);
	hBaseCollector.prepare();
	return hBaseCollector;
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openForWrite(FlowProcess<JobConf> flowProcess, OutputCollector output) throws IOException {
    return new HadoopTupleEntrySchemeCollector(flowProcess, this, output);
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openForWrite(FlowProcess<Object> flowProcess, Object output) throws IOException {
    initInnerTapIfNotSetFromFlowProcess(flowProcess);
    return actualTap.openForWrite(flowProcess, output);
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openForWrite(FlowProcess<Properties> flowProcess, Object output) throws IOException {
    return new TupleEntrySchemeCollector<Properties, Object>(flowProcess, getScheme(), output);
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public TupleEntryCollector openForWrite(FlowProcess<JobConf> flowProcess, Object output) throws IOException {
    return new TupleEntrySchemeCollector(flowProcess, getScheme(), stream.stream());
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openForWrite(FlowProcess<Properties> flowProcess, OutputStream output)
        throws IOException {
    return new TupleEntrySchemeCollector<Properties, OutputStream>(flowProcess, getScheme(), os);
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openTapForWrite(Tap tap) throws IOException {
	return tap.openForWrite( this, null ); // do not honor sinkmode as this may be opened across tasks
}
 

import cascading.tuple.TupleEntryCollector; //导入依赖的package包/类
@Override
public TupleEntryCollector openSystemIntermediateForWrite() throws IOException {
	return null; // Not required for Flink
}