flink 安裝

單機安裝flink非常簡單，在官網下載flink,並執行安裝包中bin檔案下的start-cluster.sh即可，執行成功之後訪問http://localhost:8081,會出現flink 的管理頁面

Wordcout 程式碼

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 WordCount {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment see = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<String> text = see.fromElements("That last call is " +
            "necessary to start the actual Flink job. All operations, " +
            "such as creating sources, transformations and sinks " +
            "only build up a graph of internal operations. " +
            "Only when execute() is called is this graph of operations " +
            "thrown on a cluster or executed on your local machine.");

        DataSet<Tuple2<String,Long>> result = text.flatMap(new LineSplitter())
            .groupBy(0).sum(1);

        result.writeAsText("/Users/zhumingyuan/Documents/opensource/flink-1.6.1/output/wordcount.txt");
        //result.print();

        see.execute();
    }

    public static class LineSplitter implements FlatMapFunction<String,Tuple2<String,Long>> {
        @Override
        public void flatMap(String line, Collector<Tuple2<String, Long>> collector) throws Exception {
            String[] words = line.split(" ");
            for (String word : words) {
                collector.collect(new Tuple2(word,1L));
            }
        }
    }
}
 

執行樣例

可以選擇在本地執行程式碼，也可以通過管理頁面提交jar進行執行。