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