2021年最新最全Flink系列教程_Flink快速入门(概述,安装部署)(一)(JianYi收藏)

[root@node1 bin]$ start-cluster.sh

node1:8081

[root@node1 flink]$ bin/flink run /opt/server/flink/examples/batch/WordCount.jar --input /root/words.txt

hadoop-daemon start namenode
hadoop-daemon start datanode
hadoop-daemon start secondarynamenode

/export/server/zookeeper/bin/zkServer.sh start

start-cluster.sh

/export/server/flink/bin/yarn-session.sh -n 2 -tm 800 -s 1 -d

/export/server/flink/bin/flink run  /export/server/flink/examples/batch/WordCount.jar

yarn application -kill application_1620078252622_0006

/export/server/flink/bin/flink run -m yarn-cluster -yjm 1024 -ytm 1024 /export/server/flink/examples/batch/WordCount.jar

package cn.itcast.sz22.day01;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
/**
 * Author itcast
 * Date 2021/5/4 15:25
 * 通过 flink 批处理实现 wordcount
 * 开发步骤：
 * 1. 获取环境变量
 * 2. 读取数据源
 * 3. 转换操作
 * 4. 将数据落地，打印到控制台
 * 5. 执行（流环境下）
 */
public class Wordcount {
    public static void main(String[] args) throws Exception {
        //获取环境变量
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //读取数据
        //1. 文件中读取
        //2. 获取本地的数据，开发测试用
        DataSource<String> source = env
                .fromElements("itcast hadoop spark", "itcast hadoop spark", "itcast hadoop", "itcast");
        //3. 进行转换操作
        DataSet<String> flatMapDS = source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(word);
                }
            }
        });
        //3.2 转换成元素 map
        MapOperator<String, Tuple2<String, Integer>> mapDS = flatMapDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        });
        //3.3统计
        AggregateOperator<Tuple2<String, Integer>> result = mapDS.groupBy(0).sum(1);
        //4.打印输出
        result.print();
    }
}

package cn.itcast.sz22.day01;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
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;
/**
 * Author itcast
 * Date 2021/5/4 15:55
 * 流式计算 wordcount 统计
 *  编码步骤
 *  1.准备环境-env
 *  2.准备数据-source
 *  3.处理数据-transformation
 *  4.输出结果-sink
 *  5.触发执行-execute
 */
public class Wordcount2 {
    public static void main(String[] args) throws Exception {
        //编码步骤
        //1.准备环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(1);
        //2.准备数据-source
        DataStream<String> linesDS = env
                .fromElements("itcast hadoop spark","itcast hadoop spark","itcast hadoop","itcast");
        //3.处理数据-transformation
        SingleOutputStreamOperator<String> flatMap = linesDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(word);
                }
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapDS = flatMap.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = mapDS.keyBy(t -> t.f0)
                .sum(1);
        //4.输出结果-sink
        result.print();
        //5.触发执行-execute
        env.execute();
    }
}

package cn.itcast.sz22.day01;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
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;
import java.util.Arrays;
/**
 * Author itcast
 * Date 2021/5/4 16:05
 * 通过 java 的 lambda 表达式实现 wordcount
 */
public class Wordcount3 {
    public static void main(String[] args) throws Exception {
        //1. 获取环境变量
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2. 读取数据源
        DataStreamSource<String> source = env.fromElements("itcast hadoop spark", "itcast hadoop spark", "itcast hadoop", "itcast");
        //3. 转换操作
        // void flatMap(T value, Collector<O> out)
        DataStream<String> faltMapDS = source.flatMap((String value, Collector<String> out) ->
                Arrays.stream(value.split(" "))
                        .forEach(out::collect))
                .returns(Types.STRING);
        //O map(T value)
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapDS = faltMapDS
                .map((word) -> Tuple2.of(word, 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT));
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = mapDS.keyBy(t -> t.f0).sum(1);
        //4. 将数据落地，打印到控制台
        result.print();
        //5. 执行（流环境下）
        env.execute();
    }
}