@ Override public void populateDAG ( DAG dag , Configuration conf ) { TweetsInput input = new TweetsInput ( ) ; Collector collector = new Collector ( ) ; WindowOption windowOption = new WindowOption . GlobalWindow ( ) ; ApexStream < String > tags = StreamFactory . fromInput ( input , input . output , name ( "tweetSampler" ) ) . flatMap ( new ExtractHashtags ( ) ) ; tags . window ( windowOption , new TriggerOption ( ) . accumulatingFiredPanes ( ) . withEarlyFiringsAtEvery ( 1 ) ) . addCompositeStreams ( ComputeTopCompletions . top ( 10 , true ) ) . print ( name ( "console" ) ) . endWith ( collector , collector . input , name ( "collector" ) ) . populateDag ( dag ) ; }
Populate the dag with High-Level API .