public static void main ( String [ ] args ) throws IOException { StreamingWordExtractOptions options = PipelineOptionsFactory . fromArgs ( args ) . withValidation ( ) . as ( StreamingWordExtractOptions . class ) ; options . setStreaming ( true ) ; options . setBigQuerySchema ( StringToRowConverter . getSchema ( ) ) ; ExampleUtils exampleUtils = new ExampleUtils ( options ) ; exampleUtils . setup ( ) ; Pipeline pipeline = Pipeline . create ( options ) ; String tableSpec = new StringBuilder ( ) . append ( options . getProject ( ) ) . append ( ":" ) . append ( options . getBigQueryDataset ( ) ) . append ( "." ) . append ( options . getBigQueryTable ( ) ) . toString ( ) ; pipeline . apply ( "ReadLines" , TextIO . read ( ) . from ( options . getInputFile ( ) ) ) . apply ( ParDo . of ( new ExtractWords ( ) ) ) . apply ( ParDo . of ( new Uppercase ( ) ) ) . apply ( ParDo . of ( new StringToRowConverter ( ) ) ) . apply ( BigQueryIO . writeTableRows ( ) . to ( tableSpec ) . withSchema ( StringToRowConverter . getSchema ( ) ) ) ; PipelineResult result = pipeline . run ( ) ; // ExampleUtils will try to cancel the pipeline before the program exists. exampleUtils . waitToFinish ( result ) ; }
Sets up and starts streaming pipeline .