如何构建、部署运行 Flink 程序

<!-- 基础依赖 --><dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-java</artifactId>    <version>${flink.version}</version>    <scope>provided</scope></dependency>  <!-- DataStream --><dependency>  <groupId>org.apache.flink</groupId>  <artifactId>flink-streaming-java_2.11</artifactId>  <version>1.11.3</version>  <scope>provided</scope></dependency>

curl https://flink.apache.org/q/quickstart.sh | bash -s 1.11.3 * -s 构建 flink 版本

//批处理 (DataSet) 支持离线数据public class WordCount {    public static void main(String[] args)  throws Exception{        //创建执行环境        ExecutionEnvironment env=ExecutionEnvironment.getExecutionEnvironment();
        //从文件中读取数据        String inputPath="text.txt";        DataSet<String> inputDataSet = env.readTextFile(inputPath);
        //对数据集进行处理       DataSet<Tuple2<String,Integer>>  resultSet = inputDataSet.flatMap(new MyflatMapper())                //按照第一个位置对word分组                .groupBy(0)                //将第二个位置上对数据求和                .sum(1);        resultSet.print();    }
    //自定义类实现FlatMapFunction    public static class MyflatMapper implements FlatMapFunction<String,Tuple2<String,Integer>> {
        @Override        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {            //按空格分词            String[] words=value.split(" ");            //遍历所有ord,包成二元组            for(String word:words){                out.collect(new Tuple2<>(word,1));            }        }    }}

//流处理 （DataStream）支持实时数据public class StreamWordCount {    /**     * @author ZhaoPan     * @createTime 2022/3/2     * @description     */    public static void main(String[] args) throws Exception {        //创建流处理环境        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();        //设置并行度 相当于8个线程        //env.setParallelism(2);
        //从文件中读取数据     String inputPath="text.txt";        DataStream<String> inputDataSream = env.readTextFile(inputPath);


        //基于数据流进行转换计算        DataStream<Tuple2<String, Integer>> resultStream = inputDataSream.flatMap(new WordCount.MyflatMapper())                .keyBy(0)                .sum(1);
        resultStream.print();
        //执行任务        env.execute();    }}

# jobmanager节点可用的内存大小。jobmanager.heap.size: 1024m # The heap size for the TaskManager JVM# taskmanager节点可用的内存代大小。taskmanager.heap.size: 1024m # The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline.# 每台机器可用的cpu数量taskmanager.numberOfTaskSlots: 1 # The parallelism used for programs that did not specify and other parallelism.# 默认情况下任务的并行度parallelism.default: 1 slot 和 parallelism 总结：1、slot 是静态的概念，是指 taskmanager 具有的并发执行能力2、parallelism 是动态的概念，是指程序运行实际使用的并发能力3、设置合适的 parallelism 来提高运算效率（kafka 应用一般和 partition 一一对应或成倍数关系配置）

bin/yarn-session.sh -n 2 -s 2 -jm 1024 -tm 1024 -nm test -d 参数解读：-n(--container)：TaskManager的数量。-s(--slots)：每个TaskManager的slot数量，默认一个slot一个core，默认每个taskmanager的slot的个数为1，有时可以多一些taskmanager，做冗余。-jm：JobManager的内存（单位MB)。-tm：每个taskmanager的内存（单位MB)。-nm：yarn 的appName(现在yarn的ui上的名字)。-d：后台执行。

./bin/flink run ./examples/streaming/TopSpeedWindowing.jar

./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1