flink入門：01 構建簡單運行程序

1. mac平台安裝flink（默認最新版）

brew install apache-flink

安裝結果：

Version 1.7.1, commit ID: 89eafb4

2. jdk版本，我嘗試使用了Java8和Java11，都能兼容

3. 在flink的安裝目錄下，啟動flink

目錄一般默認在/usr/local/Cellar/apache-flink/1.7.1/ （查找flink安裝目錄：find / -name flink）

./start-cluster.sh

4. 訪問localhost:8081即可進入flink的dashboard，如下所示：

 

5. 使用idea，創建一個maven項目，注冊成flink的一個job

5.1. 創建maven項目

 

5.2. 添加文件SocketTextStreamWordCount.java，代碼如下：

package com.miaoying.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class SocketTextStreamWordCount {
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("USAGE:\nSocketTextStreamWordCount <hostname> <port>");
            return;
        }

        String hostname = args[0];
        Integer port = Integer.parseInt(args[1]);

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = env.socketTextStream(hostname, port);
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = stream.flatMap(new LineSplitter())
                .keyBy(0)
                .sum(1);
        sum.print();
        env.execute("Java WordCount from SocketTextStream Excample");
    }

    public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) {
            String[] tokens = s.toLowerCase().split("\\W+");
            for (String token : tokens) {
                if (token.length() > 0) {
                    collector.collect(new Tuple2<String, Integer>(token, 1));
                }
            }
        }
    }
}

5.3. 進入項目工程目錄，打成jar包：

mvn clean package -Dmaven.skip.test=true

5.4. 打開terminal，開啟監聽端口 9000

nc -l 9000

5.5. 進入flink安裝目錄bin下執行以下命令：

flink run -c com.miaoying.flink.SocketTextStreamWordCount /Users/miaoying/Documents/program/flink/target/original-flink01-1.0-SNAPSHOT.jar 127.0.0.1 9000

輸出：

Starting execution of program

5.6. 在dashboard的running jobs中可以看到剛剛啟動的job，如下圖所示：

 

5.7. 可以在nc監聽端口中隨意輸入內容，比如123

5.8. 然后可以去flink的安裝目錄下查看日志信息，/usr/local/Cellar/apache-flink/1.7.1/libexec/log目錄下執行：

tail -f flink-miaoying-taskexecutor-0-localhost.out

如下所示：

 

本文flink程序簡單介紹就此結束。

 

 

學習資源主要來自：

https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/batch/examples.html#running-an-example

http://www.54tianzhisheng.cn/2018/09/18/flink-install