当前位置：首页 > news >正文

3.flinkDateStreamAPI介绍env与source

news 2025/8/22 0:58:43

执行环境

Flink可以在不同的环境上下文中运行.可以本地集成开发环境中运行也可以提交到远程集群环境运行.
不同的运行环境对应的flink的运行过程不同,需要首先获取flink的运行环境,才能将具体的job调度到不同的TaskManager
在flink中可以通过StreamExecutionEnvironment类获取不同的环境

自适应方式 getExecutionEnvironment
flink会根据运行的上下文自动推断出创建什么样的环境,也是开发中最常用的方式

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

创建本地环境 createLocalEnvironment
这个方法返回一个本地执行环境。可以在调用时传入一个参数，指定默认的并行度；如果
不传入，则默认并行度就是本地的 CPU 核心数。

StreamExecutionEnvironment.createLocalEnvironment();

创建远程集群运行环境 createRemoteEnvironment
这个方法返回集群执行环境。需要在调用时指定 JobManager 的主机名和端口号，并指定
要在集群中运行的 Jar 包。

        // 创建远程执行环境// job manager hostString host = "node1";// job manager portint port = 6123;// 默认并行度int parallelism = 1;// jar包存在位置String jarFiles = "hdfs://flink/data/wordCount.class";StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment.createRemoteEnvironment(host, port, parallelism, jarFiles);

自定义SourceFunction 实现SourceFunction可以通过我们自定义方式加载数据

SourceFunction 并发度只能是1
ParallelSourceFunction支持setParallelism

public class FlinkCustomSourceOperatorDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource ds = env.addSource(new MyCustomSourceOperator());ds.print();env.execute();}/*** 实现SourceFunction接口的run 方法 与 cancel*/static class MyCustomSourceOperator implements SourceFunction<Integer> {private boolean flag = true;private ThreadLocalRandom random = ThreadLocalRandom.current();/*** 数据收集方法* @param ctx* @throws Exception*/@Overridepublic void run(SourceContext<Integer> ctx) throws Exception {// flag标志位表示数据的生成是否停止while (flag) {// ctx source上下文 collect可以收集生成的数据流向下游ctx.collect(random.nextInt(3000));Thread.sleep(1000);}}/*** 任务停止方法*/@Overridepublic void cancel() {flag = false;}}
}

运行模式

flink 在1.12.0版本上统一了批处理与流处理的API,两种数据都可以使用DataStreamAPI进行处理.默认都是以STREAM流式模式进行处理
设置方式

命令行设置

bin/flink run -Dexecution.runtime-mode=BATCH

代码设置

    env.setRuntimeMode(RuntimeExecutionMode.BATCH)

推荐通过命令行模式进行设置运行模式,而通过代码硬编码的形式灵活度较差
关于批与流处理的选择
批处理会等到数据全部就位之后一次性输出结果,流式处理会一直等待数据写入来一条处理一条,在如果数据有界的情况下直接输出效率更高,如果数据无界就只能使用流式处理

最后在编写完成flink程序之后需要显示调用execute方法程序才会真正执行

Flink支持的数据类型

Flink支持大部分Java与Scala数据类型

基本数据类型及其包装类
数组类型包含基本类型数组和对象类型数组
复合数据类型
- POJO
- 元组
- 行类型ROW
辅助类型 Optional Either List Map等
泛型

元组类型和 POJO 类型最为灵活，复杂类型。而相比之下，POJO 还支持在键（key）的定义中直接使用字段名，这会让我们的代码可读性大大增加。

flink 对 POJO 类型的要求如下：

类是公共的（public）和独立的（没有非静态的内部类）
类有一个公共的无参构造方法
类中的所有字段是 public 且非 final 的；或者有一个公共的 getter 和 setter 方法，这些方法需要符合 Java bean 的命名规范

类型提示TypeHints

由于Java存在泛型擦除,还有一些lambda表达式的情况,flink无法推断出返回类型,此时可以通过类型提示在编译的时候就告诉flink泛型类型
flink提供改了TypeHints 与 Types两个类作为返回值类型提示明确告诉转换后的DataStream的数据类型

    public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();String filePath = FileUtil.getAbsolutePath("classpath:input/wordcount.txt");DataStreamSource<String> ds = env.readTextFile(filePath);ds.flatMap((String data, Collector<Tuple2<String, Integer>> collector) -> {String[] word = data.split(" ");Arrays.stream(word).forEach(w -> {collector.collect(Tuple2.of(w, 1));});})// TypeHint 或者 Types.returns(new TypeHint<Tuple2<String, Integer>>() {}).keyBy(data -> data.f0).sum(1).print();env.execute();}