Flink实战（三）开发过程中将程序提交到集群中执行

偏执的太偏执、 2023-10-11 10:03 33阅读 0赞

### 主要配置 ###

使用 `createRemoteEnvironment（）`方法即可

// 配置服务地址与当前jar所在路径
            final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("localhost", 8081, "D:\\01code\\personal\\flink\\flink-demo\\chapter1\\target\\chapter1-1.0-SNAPSHOT.jar");

### 编写程序，配置目标地址 ###

public static void main(String[] args) throws Exception { 
    
            Configuration config = new Configuration();
            config.setString("jobmanager.rpc.address", "jobmanager");
            config.setString("jobmanager.rpc.port", "6123");
    
            // 配置服务地址与当前jar所在路径
            final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("localhost", 8081, "D:\\01code\\personal\\flink\\flink-demo\\chapter1\\target\\chapter1-1.0-SNAPSHOT.jar");
    
            // 最好使用hdfs地址
            DataSource<String> text = env.readTextFile("D:\\01code\\personal\\flink\\flink-demo\\chapter1\\src\\main\\resources\\test\\hello.txt");
    
            env.setParallelism(3);
            // parse the data, group it, window it, and aggregate the counts
            text
                    .flatMap(new FlatMapFunction<String, WordWithCount>() { 
                        @Override
                        public void flatMap(String value, Collector<WordWithCount> out) { 
                            for (String word : value.split(" ")) { 
                                System.out.println(word);
                                out.collect(new WordWithCount(word, 1L));
                            }
                        }
                    })
                    .groupBy("word")
                    .reduce(new ReduceFunction<WordWithCount>() { 
                        @Override
                        public WordWithCount reduce(WordWithCount a, WordWithCount b) { 
                            return new WordWithCount(a.word, a.count + b.count);
                        }
                    }).collect()
                    .forEach(wordWithCount -> { 
                        System.out.println("show:::" + wordWithCount);
                    });
            //env.execute("file Window WordCount");
        }

--------------------

tip：

1.  注意如果配置了`setParallelism`，一定要保证服务器上的slot大于这个，可以通过`taskmanager.numberOfTaskSlots`配置，推荐配置当前服务器核心数