Getting started with Flink wordCount

Programming Model of Flink

1. Get the Flink context

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment ()

2. Load and create data

DataSet

3. Data conversion

Transformation

4. Data result storage

5. Trigger execution.

Env.execution

Here is the output of wordcount data for flink:

Import org.apache.flink.api.common.functions.FlatMapFunction

Import org.apache.flink.api.java.DataSet

Import org.apache.flink.api.java.ExecutionEnvironment

Import org.apache.flink.api.java.tuple.Tuple2

Import org.apache.flink.util.Collector

Public class FlinkMain {

@ SuppressWarnings ("serial") public static class LineSplit implements FlatMapFunction {@ SuppressWarnings ("rawtypes") @ Override / * @ param value raw data * @ param out output data * / public void flatMap (String value, Collector out) throws Exception {String [] tokens = value.split ("") For (String token: tokens) {if (tokenized null & & token.length () > 0) {Tuple2 t = new Tuple2 (token,1); out.collect (t);} public static void main (String [] args) throws Exception {/ / create flink context ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment () / / create dataset DataSet text = env.fromElements ("to be", "or no to be", "is question"); / / A pair of dataset transformations DataSet count = text.flatMap (new LineSplit ()); / / output the converted dataset (print contains env.execute execution) count.print (); System.out.println ("- -") / / A pair of data sets are grouped into statistical transformations. 0Power1 is the subscript, corresponding to the parameter count = count.groupBy (0) .sum (1) in the Tuple2 class; / / the console output dataset count.print (); System.out.println ("- -");}

}

Flink uses sql to transform data

Import java.util.ArrayList

Import java.util.List

Import org.apache.flink.api.java.DataSet

Import org.apache.flink.api.java.ExecutionEnvironment

Import org.apache.flink.table.api.Table

Import org.apache.flink.table.api.TableEnvironment

Import org.apache.flink.table.api.java.BatchTableEnvironment

Public class FlinkMain2 {

@ SuppressWarnings ({"unchecked", "rawtypes"}) public static void main (String [] args) throws Exception {/ / create flink context ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment (); BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment (env); List list = new ArrayList (); String workStr = "to be or no to be is question"; String [] tokens = workStr.split (") For (String token: tokens) {if (tokenized null & & token.length () > 0) {list.add (new WordCount (token,1));}} / / create a dataset DataSet input = env.fromCollection (list); / / register as a data table wordCount is a database table, word,frequency is a wordCount table field tEnv.registerDataSet ("wordCount", input, "word,frequency") Table table = tEnv.sqlQuery ("SELECT word, SUM (frequency) as frequency FROM wordCount GROUP BY word"); DataSet res = tEnv.toDataSet (table, WordCount.class); / / console output res.print ();} public static class WordCount {public String word; public long frequency; public WordCount () {} public WordCount (String word, long frequency) {this.word = word; this.frequency = frequency } @ Override public String toString () {return "words:" + word + ", word frequency:" + frequency;}}

}