import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
@Override
public TupleEntryIterator openForRead(FlowProcess<Properties> flowProcess, ScrollQuery input) throws IOException {
    if (input == null) {
        // get original copy
        Settings settings = CascadingUtils.addDefaultsToSettings(CascadingUtils.extractOriginalProperties(flowProcess.getConfigCopy()), tapProperties, log);

        // will be closed by the query is finished
        RestRepository client = new RestRepository(settings);
        Field mapping = client.getMapping();
        Collection<String> fields = CascadingUtils.fieldToAlias(settings, getSourceFields());

        // validate if possible
        FieldPresenceValidation validation = settings.getReadFieldExistanceValidation();
        if (validation.isRequired()) {
            MappingUtils.validateMapping(fields, mapping, validation, log);
        }

        input = QueryBuilder.query(settings).fields(StringUtils.concatenateAndUriEncode(fields,  ",")).
                build(client, new ScrollReader(new ScrollReaderConfig(new JdkValueReader(), mapping, settings)));
    }
    return new TupleEntrySchemeIterator<Properties, ScrollQuery>(flowProcess, getScheme(), input, getIdentifier());
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
/**
 * Tests the content of an output path against the given expected path.
 */
@SuppressWarnings("unchecked")
private void testPaths(String actual, String expected) throws Exception {

  Tap outputTest = new Hfs(new TextLine(), actual);
  Tap expectedTest = new Hfs(new TextLine(), expected);

  FlowProcess outputProcess = new HadoopFlowProcess(new JobConf(new Configuration()));
  FlowProcess expectedProcess = new HadoopFlowProcess(new JobConf(new Configuration()));

  TupleEntryIterator outputIterator = outputTest.openForRead(outputProcess);
  TupleEntryIterator expectedIterator = expectedTest.openForRead(expectedProcess);

  List<String> outputList = new ArrayList<>();
  while (outputIterator.hasNext()) {
    outputList.add(outputIterator.next().getTuple().getString(1));
  }

  List<String> expectedList = new ArrayList<>();
  while (expectedIterator.hasNext()) {
    expectedList.add(expectedIterator.next().getTuple().getString(1));
  }

  assertTrue(outputList.equals(expectedList));

}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
public static boolean compareTaps(final Tap source1, final Tap source2, final Configuration conf) throws IOException {
    final FlowProcess flowProcess1 = new HadoopFlowProcess(new JobConf(conf));
    source1.getScheme().retrieveSourceFields(flowProcess1, source1);
    final TupleEntryIterator iter1 = source1.openForRead(new HadoopFlowProcess(new JobConf(conf)));
    final FlowProcess flowProcess2 = new HadoopFlowProcess(new JobConf(conf));
    source2.getScheme().retrieveSourceFields(flowProcess2, source2);
    final TupleEntryIterator iter2 = source2.openForRead(new HadoopFlowProcess(new JobConf(conf)));
    if (!iter1.getFields().equals(iter2.getFields()))
        return false;
    List<Tuple> list1 = new ArrayList<Tuple>();
    while (iter1.hasNext())
        list1.add(new Tuple(iter1.next().getTuple()));
    iter1.close();
    Collections.sort(list1);
    List<Tuple> list2 = new ArrayList<Tuple>();
    while (iter2.hasNext())
        list2.add(new Tuple(iter2.next().getTuple()));
    iter2.close();
    Collections.sort(list2);
    return list1.equals(list2);
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
/**
 * Reads the {@link Tuple Tuples} from the {@link Tap} and returns them wrapped in a {@link Data} instance whose
 * {@link Fields} confirm to those supplied by {@link Tap#getSourceFields()}.
 */
Data read() throws IOException {
  TupleEntryIterator tuples = null;
  try {
    Class<?> tapConfigClass = TapTypeUtil.getTapConfigClass(source);

    if (Configuration.class.equals(tapConfigClass)) {
      tuples = getHadoopTupleEntryIterator();
    } else if (Properties.class.equals(tapConfigClass)) {
      tuples = getLocalTupleEntryIterator();
    } else {
      throw new IllegalArgumentException("Unsupported tap type: " + source.getClass());
    }
    List<Tuple> resultTuples = new ArrayList<Tuple>();
    while (tuples.hasNext()) {
      resultTuples.add(new Tuple(tuples.next().getTuple()));
    }
    return new Data(source.getSourceFields(), Collections.unmodifiableList(resultTuples));
  } finally {
    if (tuples != null) {
      tuples.close();
    }
  }
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
public void runSplunkScheme(String path, String inputData) throws IOException
{
    Properties properties = TestConfigurations.getSplunkLoginAsProperties();
    SplunkScheme inputScheme = new SplunkScheme(TestConfigurations.getSplunkSearch());
    TextLine outputScheme = new TextLine();

    SplunkTap input = new SplunkTap(properties,inputScheme);
    Hfs output = new Hfs( outputScheme, outputPath + "/quoted/" + path, SinkMode.REPLACE );

    Pipe pipe = new Pipe( "test" );
    Flow flow = new HadoopFlowConnector().connect( input, output, pipe );

    flow.complete();

    validateLength( flow, 10, 2 );

    TupleEntryIterator iterator = flow.openSource();

    // TODO: Header information not used in SplunkScheme yet
    // verifyHeader(iterator.getFields());

    verifyContent(iterator);
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
@Test
public void exerciseScheme() throws IOException {
  TupleEntryIterator iterator = tap.openForRead(flowProcess);
  while (iterator.hasNext()) {
    iterator.next();
  }
  iterator.close();
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
private TupleEntryIterator getHadoopTupleEntryIterator() throws IOException {
  @SuppressWarnings("unchecked")
  Tap<JobConf, ?, ?> hadoopTap = (Tap<JobConf, ?, ?>) source;
  JobConf conf = new JobConf();
  FlowProcess<JobConf> flowProcess = new HadoopFlowProcess(conf);
  hadoopTap.sourceConfInit(flowProcess, conf);
  return hadoopTap.openForRead(flowProcess);
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
private TupleEntryIterator getLocalTupleEntryIterator() throws IOException {
  @SuppressWarnings("unchecked")
  Tap<Properties, ?, ?> localTap = (Tap<Properties, ?, ?>) source;
  Properties properties = new Properties();
  FlowProcess<Properties> flowProcess = new LocalFlowProcess(properties);
  localTap.sourceConfInit(flowProcess, properties);
  return localTap.openForRead(flowProcess);
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
@Test
public void openForRead() throws IOException {
  TupleEntryIterator iterator = tap.openForRead(null, null);
  assertThat(iterator.hasNext(), is(true));
  assertThat(iterator.next(), is(new TupleEntry(FIELDS, TUPLE_1)));
  assertThat(iterator.hasNext(), is(true));
  assertThat(iterator.next(), is(new TupleEntry(FIELDS, TUPLE_2)));
  assertThat(iterator.hasNext(), is(false));
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
private void verifyContent(TupleEntryIterator iterator) throws IOException {

        String [] expectedRows = new String[] {
                "1,count=4",
                "2,count=3",
                "3,count=2",
                "4,count=1",
                "5,count=0"
        };
        for(String expectedRow : expectedRows)
            checkResults(iterator.next().getTuple(), expectedRow);
    }
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
public void runCSVLine( String path, String inputData) throws IOException
{
    Properties properties = new Properties();

    CSVLine inputScheme = new CSVLine();
    TextLine outputScheme = new TextLine();

    Hfs input = new Hfs( inputScheme, inputData );
    Hfs output = new Hfs( outputScheme, outputPath + "/quoted/" + path, SinkMode.REPLACE );

    Pipe pipe = new Pipe( "test" );
    Flow flow = new HadoopFlowConnector( properties ).connect( input, output, pipe );

    flow.complete();

    validateLength( flow, 4, 2 ); // The file contains 4 rows, however there are only 3 CSV rows (inc the header row)


    TupleEntryIterator iterator = flow.openSource();

    ArrayListTextWritable expected = new ArrayListTextWritable();

    expected.add(new Text("header1"));
    expected.add(new Text("header2"));
    assertEquals(expected, iterator.next().getTuple().getObject(1));

    expected.clear();
    expected.add(new Text("Column1"));
    expected.add(new Text("Column 2 using\ntwo rows"));
    assertEquals(expected, iterator.next().getTuple().getObject(1));

    expected.clear();
    expected.add(new Text("c1"));
    expected.add(new Text("c2"));
    assertEquals(expected, iterator.next().getTuple().getObject(1));
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
@Override
public TupleEntryIterator openForRead(FlowProcess<JobConf> flowProcess, RecordReader input) throws IOException {
    return new HadoopTupleEntrySchemeIterator(flowProcess, this, input);
}
 

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

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
@Override
public TupleEntryIterator openTapForRead(Tap tap) throws IOException {
	return tap.openForRead( this );
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p/>
 * Returned type is a {@link ListTupleEntryIterator}.
 */
@Override
public TupleEntryIterator openForRead(FlowProcess<? extends Properties> flowProcess, Iterator<Tuple> input)
  throws IOException {
  return new ListTupleEntryIterator(getSourceFields(), this.input);
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
/**
 * Always throws {@link UnsupportedOperationException} - this is a sink not a tap.
 *
 * @throws UnsupportedOperationException always.
 */
@Override
public TupleEntryIterator openForRead(FlowProcess<? extends Properties> flowProcess, Iterator<Tuple> input)
    throws IOException {
  throw new UnsupportedOperationException("cannot read from a " + getClass().getSimpleName());
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
@Override
public TupleEntryIterator openForRead(FlowProcess<? extends String> flowProcess, Long input) throws IOException {
  return null;
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
List<Tuple> launchFlow() throws IOException {
    File tmpDir = Files.createTempDir();
    try {
        FlowDef flowDef = FlowDef.flowDef().setName("testFlow");
        for(int sourceId=0; sourceId<sources.length; sourceId++) {
            Source source = sources[sourceId];
            File input = new File(tmpDir, "input"+sourceId+".csv");
            FileWriter writer = new FileWriter(input);
            try {
                for(Tuple tuple : source.tuples) {
                    if(source.fields.size() != tuple.size()) {
                        throw new IllegalArgumentException("Number of input fields is not the same of value of input tuple");
                    }
                    writer.write((tuple.getString(0) == null)? "" : tuple.getString(0));
                    for(int i=1; i<tuple.size(); i++) {
                        writer.write("\t");
                        writer.write((tuple.getString(i) == null)? "" : tuple.getString(i));
                    }
                    writer.write("\n");
                }
            } finally {
                writer.flush();
                writer.close();
            }
            FileTap inputTap = new FileTap(new TextDelimited(source.fields), input.getAbsolutePath());
            flowDef.addSource(source.pipe, inputTap);
        }
        File output = new File(tmpDir, "output.csv");
        FileTap outputTap = new FileTap(new TextDelimited(true, "\t"), output.getAbsolutePath());
        flowDef.addTailSink(tail, outputTap);
        Flow<?> flow = new LocalFlowConnector(new Properties()).connect(flowDef);
        flow.complete();
        
        List<Tuple> result = new ArrayList<Tuple>();
        TupleEntryIterator iterator = flow.openSink();
        while(iterator.hasNext()) {
            result.add(iterator.next().getTupleCopy());
        }
        iterator.close();
        return result;
    } finally {
        FileUtils.deleteDirectory(tmpDir);
    }
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
@Override
public TupleEntryIterator openForRead(FlowProcess<JobConf> flowProcess, RecordReader recordReader) throws IOException {
    return new HadoopTupleEntrySchemeIterator( flowProcess, this, recordReader );
}
 

import cascading.tuple.TupleEntryIterator; //导入依赖的package包/类
@Override
public TupleEntryIterator openForRead(FlowProcess<Properties> flowProcess, ScrollQuery input) throws IOException {
    if (input == null) {
        // get original copy
        Settings settings = CascadingUtils.addDefaultsToSettings(CascadingUtils.extractOriginalProperties(flowProcess.getConfigCopy()), tapProperties, log);

        // will be closed by the query is finished
        RestRepository client = new RestRepository(settings);
        MappingSet mappings = client.getMappings();
        Mapping mapping = mappings.isEmpty() ? null : mappings.getResolvedView();
        Collection<String> fields = CascadingUtils.fieldToAlias(settings, getSourceFields());

        String userFilter = settings.getReadSourceFilter();
        if (StringUtils.hasText(userFilter)){
            if (fields.isEmpty()) {
                fields = StringUtils.tokenize(userFilter, ",");
            } else {
                throw new EsHadoopIllegalStateException("User specified source filters were found [" + userFilter + "], " +
                        "but the connector is executing in a state where it has provided its own source filtering " +
                        "[" + StringUtils.concatenate(fields, ",") + "]. Please clear the user specified source fields under the " +
                        "[" + ConfigurationOptions.ES_READ_SOURCE_FILTER + "] property to continue.");
            }
        }

        // validate if possible
        FieldPresenceValidation validation = settings.getReadFieldExistanceValidation();
        if (validation.isRequired()) {
            MappingUtils.validateMapping(fields, mapping, validation, log);
        }
        
        EsMajorVersion esVersion = settings.getInternalVersionOrThrow();
        Resource read = new Resource(settings, true);
        SearchRequestBuilder queryBuilder =
                new SearchRequestBuilder(esVersion, settings.getReadMetadata() && settings.getReadMetadataVersion())
                        .types(read.type())
                        .indices(read.index())
                        .query(QueryUtils.parseQuery(settings))
                        .scroll(settings.getScrollKeepAlive())
                        .size(settings.getScrollSize())
                        .limit(settings.getScrollLimit())
                        .filters(QueryUtils.parseFilters(settings))
                        .fields(StringUtils.concatenate(fields, ","));
        input = queryBuilder.build(client, new ScrollReader(new ScrollReaderConfig(new JdkValueReader(), mapping, settings)));
    }
    return new TupleEntrySchemeIterator<Properties, ScrollQuery>(flowProcess, getScheme(), input, getIdentifier());
}