郑州市做网站,济南高端网站设计,企业取名,天眼执行环境
Flink可以在不同的环境上下文中运行.可以本地集成开发环境中运行也可以提交到远程集群环境运行. 不同的运行环境对应的flink的运行过程不同,需要首先获取flink的运行环境,才能将具体的job调度到不同的TaskManager 在flink中可以通过StreamExecutionEnvironment类获取…执行环境
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 并发度只能是1ParallelSourceFunction支持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 SourceFunctionInteger {private boolean flag true;private ThreadLocalRandom random ThreadLocalRandom.current();/*** 数据收集方法* param ctx* throws Exception*/Overridepublic void run(SourceContextInteger 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-modeBATCH代码设置 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);DataStreamSourceString ds env.readTextFile(filePath);ds.flatMap((String data, CollectorTuple2String, Integer collector) - {String[] word data.split( );Arrays.stream(word).forEach(w - {collector.collect(Tuple2.of(w, 1));});})// TypeHint 或者 Types.returns(new TypeHintTuple2String, Integer() {}).keyBy(data - data.f0).sum(1).print();env.execute();}
