Flink02--代码选择

转换算子transformation

Iterate

其他

数据源Source

本地web ui环境

可以进行循环的map计算

Source

Mysql数据库键表对应source.addSink(        JdbcSink.sink(                \

StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

结束时调用

common基础的

environment.execute();

流式计算运行环境

Kafka

Connectors连接器

socketTextStream - 从套接字读取.DataStreamSource<String> lineStream =environment.socketTextStream(\"localhost\

close当这个函数对象运行完毕之后，默认执行一次类似于JUnit的 After注解程序员在代码执行完毕后想操作某些东西优先考虑这个方法例如:关闭数据库连接

Common

source.writeAsText(\"out/test_\"+System.currentTimeMillis());

mysql...

Union

setRuntimeContext代码运行的时候会自动的设置运行的上下文环境程序员一般情况下不需要去操作这个方法

执行环境

基于套接字

Common基础的

开始时调用

source.print(\"输出到控制台\");

jdbc

getRuntimeContext这个就是为了获取系统的上下文环境那么对应函数的“主”方法就是我们操作的场景例如: MapFunction -->map

//数据源DataStreamSource<String> source = environment.fromElements(\

运行环境

source.writeToSocket(\"192.168.1.101\

Socket

readTextFile(path) - 读取文本文件DataSet<String> source =environment.readTextFile(\"data/wordcount.txt\");

输出sink

JDBC

将数据进行加密存储

kafka

代码分类

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

RichParallelSourceFunction 和 ParallelSourceFunctionRichParallelSourceFunction 可以有更多的方法例如: getRuntimeContext().getNumberOfParallelSubtasks(); (获取总的并行数)getRuntimeContext().getIndexOfThisSubtask();(获取当前的子任务索引)

控制台

Rich方法

sink

1.实现接口 org.apache.flink.streaming.api.functions.source.ParallelSourceFunction2.重写run方法，开始解析数据3.environment.addSource 访问指定的数据源 YjxxtCustomSource

StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();

Custom自定义

https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/datastream/kafka///1.sourceKafkaSource<String> sourceSetting = KafkaSource.<String>builder()        .setBootstrapServers(\

批处理运行环境

基于文件

合并流

相同的数据会被保留.

Filetext/CSV

https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/datastream/kafka/#kafka-sink//配置卡夫卡KafkaSink<String> kafkaSinkSetting = KafkaSink.<String>builder()        .setBootstrapServers(\

//输出source.addSink(new YjxxtCustomSink(\"data/sink\" + System.currentTimeMillis())).setParallelism(1);//运行环境environment.execute();

map:转换flatMap:压缩fifter:过滤sum min max minBy maxBy :滚动聚合reduce:自定义的滚动聚合

基于集合

transformation

问题

Rich富有的

getNumberOfParallelSubtasksgetIndexOfThisSubtask

Connet

open当这个函数对象被调用的时候，默认执行一次类似于JUnit的 Before注解程序员有某些想法的时候，优先考虑调用这个方法