import cascading.scheme.hadoop.TextLine; //导入依赖的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.scheme.hadoop.TextLine; //导入依赖的package包/类
public static void main(String [] args) {

        Properties properties = new Properties();

        properties.put(SplunkConf.SPLUNK_USERNAME, "admin");
        properties.put(SplunkConf.SPLUNK_PASSWORD, "changeIt");
        properties.put(SplunkConf.SPLUNK_HOST, "localhost");
        properties.put(SplunkConf.SPLUNK_PORT, "9050");

        SplunkDataQuery splunkSearch = new SplunkDataQuery();
        SplunkScheme inputScheme = new SplunkScheme(splunkSearch);
        SplunkTap input = new SplunkTap(properties,inputScheme);

        TextLine outputScheme = new TextLine();
        Hfs output = new Hfs( outputScheme, PATH_TO_OUTPUT, SinkMode.REPLACE );

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

        flow.complete();
    }
 

import cascading.scheme.hadoop.TextLine; //导入依赖的package包/类
public Flow runCascadingJob( Path inputPath, Path outputPath) throws IOException
{
    Properties properties = new Properties();

    ShuttlCsv inputScheme = new ShuttlCsv(new SplunkDataQuery());
    TextLine outputScheme = new TextLine();

    Hfs input = new Hfs(inputScheme,inputPath.toString());
    Hfs output = new Hfs(outputScheme,outputPath.toString(),SinkMode.REPLACE);

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

    flow.complete();

    return flow;
}
 

import cascading.scheme.hadoop.TextLine; //导入依赖的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.scheme.hadoop.TextLine; //导入依赖的package包/类
/**
 * Reads the header record from the source file.
 */
@SuppressWarnings("unchecked")
private CSVRecord getHeaderRecord(FlowProcess<JobConf> flowProcess, Tap tap) {
  Tap textLine = new Hfs(new TextLine(new Fields("line")), tap.getFullIdentifier(flowProcess.getConfigCopy()));

  try (TupleEntryIterator iterator = textLine.openForRead(flowProcess)) {
    String line = iterator.next().getTuple().getString(0);
    boolean skipHeaderRecord = format.getSkipHeaderRecord();
    CSVRecord headerRecord = CSVParser.parse(line, format.withSkipHeaderRecord(false)).iterator().next();
    format.withSkipHeaderRecord(skipHeaderRecord);
    return headerRecord;
  } catch (IOException e) {
    throw new TapException(e);
  }
}
 

import cascading.scheme.hadoop.TextLine; //导入依赖的package包/类
@Override
public Tap getTextFile(Fields sourceFields, Fields sinkFields, String filename, SinkMode mode) {
	if( sourceFields == null ) {
			return new Hfs(new TextLine(), filename, mode);
	}

	return new Hfs( new TextLine( sourceFields, sinkFields ), filename, mode );
}
 

import cascading.scheme.hadoop.TextLine; //导入依赖的package包/类
public OpenCsvScheme(final Fields fields, final boolean hasHeader, final char separator, final char quote, final char escape, 
                     final boolean strict, final String charsetName) {
    super(TextLine.Compress.DEFAULT);
    setCharsetName(charsetName);
    setSinkFields(fields);
    setSourceFields(fields);
    
    this.hasHeader = hasHeader;
    this.separator = separator;
    this.quote = quote;
    this.escape = escape;
    this.strict = strict;
    this.charsetName = charsetName != null ? charsetName : "UTF8";
}
 

import cascading.scheme.hadoop.TextLine; //导入依赖的package包/类
public static void main(String [] args) {

        ShuttlCsv csv = new ShuttlCsv();
        Hfs input = new Hfs(csv,PATH_TO_SHUTTL_ARCHIVE);

        TextLine outputScheme = new TextLine();
        Hfs output = new Hfs(outputScheme, PATH_TO_OUTPUT);
        Pipe pipe = new Pipe( "test" );

        Properties properties = new Properties();
        Flow flow = new HadoopFlowConnector( properties ).connect( input, output, pipe );

        flow.complete();
    }
 

import cascading.scheme.hadoop.TextLine; //导入依赖的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));
}