Mac上搭建fliink

Mac快速搭建flink环境

1、安装flink

（1）、Mac上安装flink很方便。推荐通过 homebrew 来安装：

 brew install apache-flink

（2）、检查安装

flink --version

结果：

Version: 1.9.1, Commit ID: 4d56de8

（3）、启动 flink

/usr/local/Cellar/apache-flink/1.9.1/libexec/bin ./start-cluster.sh

结果：

Starting cluster.
Starting standalonesession daemon on host PENGWEIXU-MB1.
Starting taskexecutor daemon on host PENGWEIXU-MB1.

（4）、紧接着, 接入web页面 http://localhost:8081/
查看服务

2、编写demo

（1）新建一个maven项目

（2）创建SocketTextStreamWordCount文件

package com.flink.demo;

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]);


        // set up the streaming execution environment
        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 Example");
    }

    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));
                }
            }
        }
    }
}

（3）进入工程目录, 使用以下命令打包

mvn clean package -Dmaven.test.skip=true

（4）然后开启监听 9000端口：

nc -l 9000

（5）进入flink 安装所在目录 bin下执行如下命令跑程序

flink run -c com.flink.demo.SocketTextStreamWordCount /Users/pengwei/working/git/flinkhello/target/original-flink-hello-1.0-SNAPSHOT.jar 127.0.0.1 9000

注意换成自己本地所在的项目路径

执行后结果

Starting execution of program

也可以从web页面 http://localhost:8081/
看到已有任务启动

（6）然后在监听 9000端口中输入text。如：

nc -l 9000

输入

通过 tail 命令看一下输出的 log 文件，来观察统计结果。进入目录 apache-flink/1.9.1/libexec/log，执行以下命令:

tail -300f flink-pengwei-taskexecutor-0-PENGWEIXU-MB1.out

注意切换成自己的路径、目录