问TextIO.read().watchForNewFiles()防止写入BigQuery

EN

public static void main(String[] args) {

    Pipeline p = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class));
    TableReference tableRef = new TableReference();
    tableRef.setProjectId(PROJECT_ID);
    tableRef.setDatasetId(DATASET_ID);
    tableRef.setTableId(TABLE_ID);
    //Pipeline p = Pipeline.create(PipelineOptionsFactory.as(Options.class));

    // Read files as they arrive in GS
    p.apply("ReadFile", TextIO.read()
        .from("gs://mybucket/*.csv")
        .watchForNewFiles(
            // Check for new files every 30 seconds
            Duration.standardSeconds(30),
            // Never stop checking for new files
            Watch.Growth.<String>never()
        )
    )
    .apply(ParDo.of(new DoFn<String, Segment>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            String[] items = c.element().split(",");

            if (items[0].startsWith("_", 1)) {
                // Skip header (the header is starting with _comment)
                LOG.info("Skipped header");
                return;
            }

            Segment segment = new Segment(items);
            c.output(segment);
        }
    }))
    .apply(ParDo.of(new FormatSegment()))
    .apply(BigQueryIO.writeTableRows()
        .to(tableRef)
        .withSchema(FormatSegment.getSchema())
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

    // Run the pipeline.
    p.run();
}