java 代码

package day1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class WordCountExample4socket {
    static final String KEY_HOSTNAME = "hostname";
    static final String KEY_PORT = "port";

    public static void main(String[] args) throws Exception {
        ParameterTool pt=ParameterTool.fromArgs(args);

        if(!pt.has(KEY_HOSTNAME)){
//            System.out.println("请输入 hostname");
//            return;
            throw new Exception("请输入 hostname");
        }

        if(!pt.has(KEY_PORT)){
//            System.out.println("请输入 port");
//            return;
            throw new Exception("请输入 port");
        }

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置模式，默认为流模式
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH);//批模式
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);//流模式
//        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断


        DataStream<String> text = env.socketTextStream(pt.get(KEY_HOSTNAME), pt.getInt(KEY_PORT));

        // 转换操作
        SingleOutputStreamOperator<Tuple2<String, Integer>> soso = text
                .flatMap(
                        (String line, Collector<Tuple2<String, Integer>> out) -> {
                            String[] arr = line.split(" ");
                            for (String word : arr) {
                                out.collect(Tuple2.of(word, 1));
                            }
                        }
                )
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy( value -> value.f0 )
                .sum(1); //对哪一个元素进行计算

        // 是 sink 操作
        soso.print();

        env.execute("my_word_count");
    }

}

idea 打包

执行 maven 的 package 命令，包名改为：test_flink.jar

上传

将 test_flink.jar 包上传到 hadoop1 的 /program 目录下

启动 tcp 服务器

nc -lk 9090

执行

flink run -m yarn-cluster -c day1.WordCountExample4socket /program/std_flink-1.0-SNAPSHOT.jar  --hostname hadoop1 --port 9090

结果如下：

测试

在 nc 服务器中输入：

lilei hanmeimei
李雷 韩梅梅 李雷

在 flink 控制台中没有打印，因为是集群运行

查看打印结果-flink web 方式

通过 上面 web网页：http://hadoop2:43659 查看，如下图：

原文出处：/show_1IX4WimeL0dH.html