import cascading.tap.SinkMode; //导入依赖的package包/类
@Test(expected = FlowException.class)
public void testWhenExtraColumnsStrict() throws Exception {
  String sourcePath = "src/test/resources/input/with-extra-columns.txt";
  String sinkPath = "src/test/resources/input/sink-with-headers";

  FlowConnector connector = new Hadoop2MR1FlowConnector();
  CSVFormat sourceFormat = CSVFormat.newFormat('\t')
    .withHeader("id", "first name", "last name", "city", "zip")
    .withQuote('"')
    .withEscape('\\')
    .withRecordSeparator('\n');

  CSVFormat sinkFormat = CSVFormat.newFormat('\t')
    .withEscape('\\')
    .withRecordSeparator('\n');

  Tap source = new Hfs(new CsvScheme(sourceFormat, true), sourcePath);
  Tap sink = new Hfs(new CsvScheme(sinkFormat), sinkPath, SinkMode.REPLACE);

  Pipe pipe = new Pipe("pipe");

  connector.connect(source, sink, pipe).complete();
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
@Test(expected = FlowException.class)
public void testWhenExtraColumnsStrictNoHeaders() throws Exception {
  String sourcePath = "src/test/resources/input/with-extra-columns-no-header.txt";
  String sinkPath = "src/test/resources/input/sink-no-headers";

  FlowConnector connector = new Hadoop2MR1FlowConnector();
  CSVFormat sourceFormat = CSVFormat.newFormat('\t')
    .withQuote('"')
    .withEscape('\\')
    .withRecordSeparator('\n');

  CSVFormat sinkFormat = CSVFormat.newFormat('\t')
    .withEscape('\\')
    .withRecordSeparator('\n');

  Tap source = new Hfs(new CsvScheme(sourceFormat, true), sourcePath);
  Tap sink = new Hfs(new CsvScheme(sinkFormat), sinkPath, SinkMode.REPLACE);

  Pipe pipe = new Pipe("pipe");

  connector.connect(source, sink, pipe).complete();
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
@Test
public void testWithHeader() throws Exception {
    final String inputFile = "quoted_header.csv";
    final String outputDir = "quoted_header";
    final String compareFile = "quoted_header.csv";
    final Properties props = new Properties();
    final Configuration conf = new Configuration();

    final Tap source = new Hfs(new OpenCsvScheme(), DATA_DIR + "/" + inputFile, SinkMode.KEEP);
    final Tap sink = new Hfs(new OpenCsvScheme(), TMP_DIR + "/" + outputDir, SinkMode.REPLACE);
    final Pipe pipe = new Each(new Pipe("test"), new Debug());
    new HadoopFlowConnector(props).connect(source, sink, pipe).complete();
    
    final Tap compare = new Hfs(new OpenCsvScheme(), COMPARE_DIR + "/" + compareFile, SinkMode.KEEP);
    assertTrue(compareTaps(sink, compare, conf) == true);
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
@Test
public void testHeaderless() throws Exception {
    final String inputFile = "quoted_headerless.csv";
    final String outputDir = "quoted_headerless";
    final String compareFile = "quoted_headerless.csv";
    final Properties props = new Properties();
    final Configuration conf = new Configuration();

    final Tap source = new Hfs(new OpenCsvScheme(new Fields("id", "product", "descr")), DATA_DIR + "/" + inputFile, SinkMode.KEEP);
    final Tap sink = new Hfs(new OpenCsvScheme(new Fields("id", "product", "descr")), TMP_DIR + "/" + outputDir, SinkMode.REPLACE);
    final Pipe pipe = new Each(new Pipe("test"), new Debug());
    new HadoopFlowConnector(props).connect(source, sink, pipe).complete();

    final Tap compare = new Hfs(new OpenCsvScheme(new Fields("id", "product", "descr")), COMPARE_DIR + "/" + compareFile, SinkMode.KEEP);
    assertTrue(compareTaps(sink, compare, conf) == true);
}
 

import cascading.tap.SinkMode; //导入依赖的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.tap.SinkMode; //导入依赖的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.tap.SinkMode; //导入依赖的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.tap.SinkMode; //导入依赖的package包/类
public Tap createTap(Scheme scheme, String path, SinkMode sinkMode, Properties properties) {
    if (!(scheme instanceof EsScheme)) {
        throw new EsHadoopIllegalArgumentException("Unknown scheme; expected " + EsScheme.class.getName());
    }

    String host = properties.getProperty("host");
    String portString = properties.getProperty("port");
    int port = (StringUtils.hasText(portString) ? Integer.parseInt(portString) : -1);
    String query = properties.getProperty("query");

    return CascadingUtils.hadoopTap(host, port, path, query, ((EsScheme) scheme).fields, properties);
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
public EsTap(String host, int port, String resource, String query, Fields fields, Properties tapSettings) {
    super(null, SinkMode.UPDATE);
    this.resource = resource;
    this.query = query;
    this.host = host;
    this.port = port;
    this.fields = fields;
    this.props = tapSettings;
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
@Test
public void testCreateTap() {
    Fields fl = new Fields();
    Properties props = new Properties();

    Scheme scheme = factory.createScheme(fl, props);
    Tap tap = factory.createTap(scheme, "somePath", SinkMode.KEEP, props);
    assertThat(tap, notNullValue());
    assertThat(tap.getClass().getName(), containsString("HadoopTap"));
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
/**
 * Tests if subset of input fields are provided, properly outputs only that subset.
 */
@Test
public void fieldsIncludedButNotMatchLengthTest() throws Exception {

  String sourcePath = "src/test/resources/input/with-headers.txt";
  String sinkPath = "src/test/resources/output/sink-with-headers";
  String expectedPath = "src/test/resources/expected/sink-with-headers-id-only.txt";

  FlowConnector connector = new Hadoop2MR1FlowConnector();
  CSVFormat sourceFormat = CSVFormat.newFormat(',')
    .withHeader("id", "first name", "last name")
    .withQuote('"')
    .withEscape('\\')
    .withRecordSeparator('\n');

  CSVFormat sinkFormat = CSVFormat.newFormat('\t')
    .withSkipHeaderRecord()
    .withEscape('\\')
    .withRecordSeparator('\n');

  Fields sourceFields = new Fields("id");
  Tap source = new Hfs(new CsvScheme(sourceFields, sourceFormat), sourcePath);
  Tap sink = new Hfs(new CsvScheme(sinkFormat), sinkPath, SinkMode.REPLACE);
  Pipe pipe = new Pipe("pipe");

  connector.connect(source, sink, pipe).complete();

  testPaths(sinkPath, expectedPath);

}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
@Test
public void testWhenExtraColumnsNotStrict() throws Exception {
  String sourcePath = "src/test/resources/input/with-extra-columns.txt";
  String sinkPath = "src/test/resources/input/sink-with-headers";
  String expectedPath = "src/test/resources/expected/with-extra-columns-no-strict.txt";
  String trapPath = "src/test/resources/input/trap-sink-with-headers";
  String expectedTrapPath = "src/test/resources/expected/trap-with-extra-columns-no-strict.txt";

  FlowConnector connector = new Hadoop2MR1FlowConnector();
  CSVFormat sourceFormat = CSVFormat.newFormat('\t')
    .withQuote('"')
    .withHeader("id", "first name", "last name", "city", "zip")
    .withEscape('\\')
    .withRecordSeparator('\n');

  CSVFormat sinkFormat = CSVFormat.newFormat('\t')
    .withSkipHeaderRecord()
    .withEscape('\\')
    .withRecordSeparator('\n');

  Tap source = new Hfs(new CsvScheme(sourceFormat, false), sourcePath);
  Tap sink = new Hfs(new CsvScheme(sinkFormat), sinkPath, SinkMode.REPLACE);
  Tap trap = new Hfs(new TextDelimited(true, "\t"), trapPath, SinkMode.REPLACE);

  Pipe pipe = new Pipe("pipe");

  connector.connect("extra-columns-not-strict", source, sink, trap, pipe).complete();

  testPaths(sinkPath, expectedPath);
  testPaths(trapPath, expectedTrapPath);
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
@Test
public void testWhenExtraColumnsNotStrictNoHeaders() throws Exception {
  String sourcePath = "src/test/resources/input/with-extra-columns-no-header.txt";
  String sinkPath = "src/test/resources/input/sink-no-headers";
  String trapPath = "src/test/resources/input/trap-no-headers";
  String expectedPath = "src/test/resources/expected/with-extra-columns-no-strict-no-header.txt";
  String expectedTrapPath = "src/test/resources/expected/trap-with-extra-columns-no-strict-no-header.txt";

  FlowConnector connector = new Hadoop2MR1FlowConnector();
  CSVFormat sourceFormat = CSVFormat.newFormat('\t')
    .withQuote('"')
    .withEscape('\\')
    .withRecordSeparator('\n');

  CSVFormat sinkFormat = CSVFormat.newFormat('\t')
    .withEscape('\\')
    .withRecordSeparator('\n');

  Tap source = new Hfs(new CsvScheme(sourceFormat, false), sourcePath);
  Tap sink = new Hfs(new CsvScheme(sinkFormat), sinkPath, SinkMode.REPLACE);
  Tap trap = new Hfs(new TextDelimited(false, "\t"), trapPath, SinkMode.REPLACE);

  Pipe pipe = new Pipe("pipe");

  connector.connect("test-extra-columns-no-header", source, sink, trap, pipe).complete();
  testPaths(sinkPath, expectedPath);
  testPaths(trapPath, expectedTrapPath);
}
 

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

		if (args.length < 2) {
			throw new IllegalArgumentException("Please specify input and ouput paths as arguments.");
		}

		Fields token = new Fields( "token", String.class );
		Fields text = new Fields( "text" );
		RegexSplitGenerator splitter = new RegexSplitGenerator( token, "\\s+" );
		// only returns "token"
		Pipe docPipe = new Each( "token", text, splitter, Fields.RESULTS );

		Pipe wcPipe = new Pipe( "wc", docPipe );
		wcPipe = new AggregateBy( wcPipe, token, new CountBy(new Fields("count")));

		Tap inTap = new Hfs(new TextDelimited(text, "\n" ), args[0]);
		Tap outTap = new Hfs(new TextDelimited(false, "\n"), args[1], SinkMode.REPLACE);

		FlowDef flowDef = FlowDef.flowDef().setName( "wc" )
				.addSource( docPipe, inTap )
				.addTailSink( wcPipe, outTap );

		FlowConnector flowConnector = new FlinkConnector();

		Flow wcFlow = flowConnector.connect( flowDef );

		wcFlow.complete();
	}
 

import cascading.tap.SinkMode; //导入依赖的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.tap.SinkMode; //导入依赖的package包/类
protected BaseTemplateTap(Tap parent, Granularity input,
    Granularity output, SinkMode sinkMode) {
  super(new TemplateScheme(parent.getScheme(), getPathFields(input,
      output)), sinkMode);
  this.parent = parent;
  this.pathTemplate = getPathTemplate(input, output);
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
@ConstructorProperties({ "parent", "input", "output", "sinkMode" })
public HfsTemplateTap(Hfs parent, Granularity input, Granularity output,
    SinkMode sinkMode) {
  super(parent,  input, output, sinkMode);
  this.input = input;
  this.output = output;
}
 

import cascading.tap.SinkMode; //导入依赖的package包/类
@ConstructorProperties({ "parent", "pathTemplate", "pathFields", "sinkMode" })
public LocalTemplateTap(FileTap parent, Granularity input, Granularity output
    , SinkMode sinkMode) {
  super(parent, input,output, sinkMode);
  this.input = input;
  this.output = output;
}
 

import cascading.tap.SinkMode; //导入依赖的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.tap.SinkMode; //导入依赖的package包/类
/**
 * Constructor JDBCTap creates a new JDBCTap instance.
 *
 * @param connectionUrl   of type String
 * @param username        of type String
 * @param password        of type String
 * @param driverClassName of type String
 * @param tableDesc       of type TableDesc
 * @param scheme          of type JDBCScheme
 * @param sinkMode        of type SinkMode
 */
public JDBCTap( String connectionUrl, String username, String password, String driverClassName, TableDesc tableDesc, JDBCScheme scheme, SinkMode sinkMode ) {
    super( scheme, sinkMode );
    this.connectionUrl = connectionUrl;
    this.username = username;
    this.password = password;
    this.driverClassName = driverClassName;
    this.tableDesc = tableDesc;

    if( tableDesc.getColumnDefs() == null && sinkMode != SinkMode.UPDATE )
        throw new IllegalArgumentException( "cannot have sink mode REPLACE or KEEP without TableDesc column defs, use UPDATE mode" );

    if( sinkMode != SinkMode.UPDATE )
        LOG.warn( "using sink mode: {}, consider UPDATE to prevent DROP TABLE from being called during Flow or Cascade setup", sinkMode );
}