如何做微网站,网站和网店的区别,官方微信公众平台,广东手机网站建设品牌1、Flink 的四大特征#xff08;基石#xff09; 2、Flink 中都有哪些 Source#xff0c;哪些 Sink#xff0c;哪些算子#xff08;方法#xff09;
预定义Source
基于本地集合的source#xff08;Collection-based-source#xff09;
基于文件的source#xff08;…1、Flink 的四大特征基石 2、Flink 中都有哪些 Source哪些 Sink哪些算子方法
预定义Source
基于本地集合的sourceCollection-based-source
基于文件的sourceFile-based-source
基于网络套接字socketTextStream
自定义Source
SourceFunction:非并行数据源(并行度只能1) --接口
RichSourceFunction:多功能非并行数据源(并行度只能1) --类
ParallelSourceFunction:并行数据源(并行度能够1) --接口
RichParallelSourceFunction:多功能并行数据源(并行度能够1) --类 【建议使用的】 3、什么是侧道输出流有什么用途
侧输出-SideOutput
Flink 通过watermark在短时间内允许了乱序到来的数据
通过延迟数据处理机制可以处理长期迟到的数据。
但总有那么些数据来的晚的太久了。允许迟到1天的设置它迟到了2天才来。
对于这样的迟到数据水印无能为力设置allowedLateness也无能为力那对于这样的数据Flink就只能任其丢掉了吗
不会Flink的两个迟到机制尽量确保了数据不会错过了属于他们的窗口但是真的迟到太久了Flink也有一个机制将这些数据收集起来
保存成为一个DataStream然后交由开发人员自行处理。
那么这个机制就叫做侧输出机制(Side Output)
4、Flink 中两个流如何合并为一个流
Union
union可以合并多个同类型的流
将多个DataStream 合并成一个DataStream
【注意】union合并的DataStream的类型必须是一致的
connect
connect可以连接2个不同类型的流(最后需要处理后再输出)
DataStream,DataStream → ConnectedStreams连接两个保持他们类型的数据流两个数据流被 Connect 之后只是被放在了一个同一个流中内部依然保持各自的数据和形式不发生任何变化【一国两制】两个流相互独立, 作为对比Union后是真的变成一个流了。
和union类似但是connect只能连接两个流两个流之间的数据类型可以不同对两个流的数据可以分别应用不同的处理逻辑.
5、Flink 中两个流如何 join
Join 算子提供的语义为 “Window join”即按照指定字段和滚动/滑动/会话窗口进行内连接(InnerJoin)。Join 将有相同 Key 并且位于同一窗口中的两条流的元素进行关联。
Join 可以支持处理时间和事件时间两种时间特征。
1.1 滚动窗口Join
当在滚动窗口上进行 Join 时所有有相同 Key 并且位于同一滚动窗口中的两条流的元素两两组合进行关联并最终传递到 JoinFunction 或 FlatJoinFunction 进行处理。 如上图所示我们定义了一个大小为 2 秒的滚动窗口最终产生 [0,1][2,3]… 这种形式的数据。上图显示了每个窗口中橘色流和绿色流的所有元素成对组合。需要注意的是在滚动窗口 [6,7] 中由于绿色流中不存在要与橘色流中元素 6、7 相关联的元素因此该窗口不会输出任何内容。
1.2 滑动窗口Join
当在滑动窗口上进行 Join 时所有有相同 Key 并且位于同一滑动窗口中的两条流的元素两两组合进行关联并最终传递到 JoinFunction 进行处理。 如上图所示我们定义了一个窗口大小为 2 秒、滑动步长为 1 秒的滑动窗口。需要注意的是一个元素可能会落在不同的窗口中因此会在不同窗口中发生关联例如绿色流中的0元素。当滑动窗口中一个流的元素在另一个流中没有相对应的元素则不会输出该元素。
6、Flink 中都有哪些 window什么是滑动滚动窗口
Window可以分成两类
CountWindow按照指定的数据条数生成一个Window与时间无关。
滚动计数窗口每隔N条数据统计前N条数据
滑动计数窗口每隔N条数据统计前M条数据
TimeWindow按照时间生成Window。
滚动时间窗口每隔N时间统计前N时间范围内的数据窗口长度N滑动距离N
滑动时间窗口每隔N时间统计前M时间范围内的数据窗口长度M滑动距离N
会话窗口按照会话划定的窗口 7、flink 中都有哪些时间语义对于 event_time 中数据迟到的处理数据乱序
EventTime:事件(数据)时间,是事件/数据真真正正发生时/产生时的时间。
IngestionTime:摄入时间,是事件/数据到达流处理系统的时间。
ProcessingTime:处理时间,是事件/数据被处理/计算时的系统的时间。
迟到处理
水印对于迟到数据不长
allowedLateness: 迟到时间很长
侧道输出对于迟到时间特别长。
8、flink 中的状态指的是什么有哪些状态你使用过哪些状态哪个项目使用到了状态
有状态计算和无状态计算
无状态计算:不需要考虑历史数据, 相同的输入,得到相同的输出!如:map, 将每个单词记为1, 进来一个hello, 得到(hello,1),再进来一个hello,得到的还是(hello,1)有状态计算:需要考虑历史数据, 相同的输入,可能会得到不同的输出! 如:sum/reduce/maxBy, 对单词按照key分组聚合,进来一个(hello,1),得到(hello,1), 再进来一个(hello,1), 得到的结果为(hello,2)
注意: Flink默认已经支持了无状态和有状态计算!
例如WordCount代码:已经做好了状态维护, 输入hello,输出(hello,1),再输入hello,输出(hello,2)。
Flink有两种基本类型的状态托管状态Managed State和原生状态Raw State。
两者的区别Managed State是由Flink管理的Flink帮忙存储、恢复和优化Raw State是开发者自己管理的需要自己序列化。
托管状态 - KeyedState ( 在keyBy之后可以使用状态 ) - ValueState (存储一个值) - ListState (存储多个值) - MapState (存储key-value) - OperatorState ( 没有keyBy的情况下也可以使用 ) [不用] - 原生状态 (不用)
9、flink 中 checkpoint 是什么如何设置。
Checkpoint:快照点, 是Flink中所有有状态的Operator在某一个时刻的State快照信息/存档信息。
一句话概括: Checkpoint就是State的快照。
可使用以下方法来设置
package com.bigdata.day06;import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** 基本功能:* program:FlinkDemo* author: 闫哥* create:2023-11-24 09:18:30**/
public class _01CheckPointDemo {public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);// 在windows运行将数据提交hdfs,会出现权限问题使用这个语句解决。System.setProperty(HADOOP_USER_NAME, root);// 在这个基础之上添加快照// 第一句开启快照每隔1s保存一次快照env.enableCheckpointing(1000);// 第二句设置快照保存的位置env.setStateBackend(new FsStateBackend(hdfs://bigdata01:9820/flink/checkpoint));// 第三句 通过webui的cancel按钮取消flink的job时不删除HDFS的checkpoint目录env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//2. source-加载数据DataStreamSourceString dataStreamSource env.socketTextStream(localhost, 9999);SingleOutputStreamOperatorTuple2String, Integer mapStream dataStreamSource.map(new MapFunctionString, Tuple2String, Integer() {Overridepublic Tuple2String, Integer map(String s) throws Exception {String[] arr s.split(,);return Tuple2.of(arr[0], Integer.valueOf(arr[1]));}});//3. transformation-数据处理转换SingleOutputStreamOperatorTuple2String, Integer result mapStream.keyBy(0).sum(1);result.print();//4. sink-数据输出//5. execute-执行env.execute();}
}
10、flink 中的重启策略 流式计算中的重启策略
重启策略的意义流式数据是不可能停止的假如有一条错误数据导致程序直接退出后面的大量数据是会丢失的对公司来讲意义是重大的损失是惨重的。
重启策略是一个单独的策略如果你配置了 checkpoint 含有重启策略的如果你没有 checkpoint 也可以自行配置重启策略总之重启策略和 checkpoint 没有必然联系。
注意此时如果有checkpoint ,是不会出现异常的需要将checkpoint的代码关闭再重启程序。会发现打印了异常那为什么checkpoint的时候不打印因为并没有log4j的配置文件需要搞一个这样的配置文件才行。
11、什么是维表 join如何实现你在哪个项目中使用过维表 join
所谓的维表Join: 进入Flink的数据需要关联另外一些存储设备的数据才能计算出来结果那么存储在外部设备上的表称之为维表可能存储在mysql也可能存储在hbase 等。
实现
通过定义一个类实现RichMapFunction在open()中读取维表数据加载到内存中在kafka流map()方法中与维表数据进行关联。
RichMapFunction中open方法里加载维表数据到内存的方式特点如下
优点实现简单缺点因为数据存于内存所以只适合小数据量并且维表数据更新频率不高的情况下。虽然可以在open中定义一个定时器定时更新维表但是还是存在维表更新不及时的情况。另外维表是变化慢不是一直不变的只是变化比较缓慢而已。
以前的方式是将维表数据存储在Redis、HBase、MySQL等外部存储中实时流在关联维表数据的时候实时去外部存储中查询这种方式特点如下
优点维度数据量不受内存限制可以存储很大的数据量。缺点因为维表数据在外部存储中读取速度受制于外部存储的读取速度另外维表的同步也有延迟。
使用cache来减轻访问压力
可以使用缓存来存储一部分常访问的维表数据以减少访问外部系统的次数比如使用Guava Cache。维表一般的特点是变化比较慢。在智慧城市项12目使用过。用它来存储一些预热的数据在内存中方便取出。
12、flinksql 如何读取 kafka 或者 mysql 的数据。
可通过以下代码直接实现 import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;/*** 基本功能:* program:FlinkDemo* author: 闫哥* create:2023-11-28 11:00:51**/
public class _02KafkaConnectorDemo {public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tEnv StreamTableEnvironment.create(env);// 如果是建表语句executeSql 这个返回值是TableResult// 如果是查询语句sqlQuery 这个返回的是Table (有用)// 新建一个表用于存储 kafka消息TableResult tableResult tEnv.executeSql(CREATE TABLE table1 (\n user_id int,\n page_id int,\n status STRING\n ) WITH (\n connector kafka,\n topic topic1,\n properties.bootstrap.servers bigdata01:9092,\n properties.group.id testGroup,\n scan.startup.mode latest-offset,\n format json\n ));// 新建一个表用于存储kafka中的topic2中的数据tEnv.executeSql(CREATE TABLE table2 (\n user_id int,\n page_id int,\n status STRING\n ) WITH (\n connector kafka,\n topic topic2,\n properties.bootstrap.servers bigdata01:9092,\n format json\n ));tEnv.executeSql(insert into table2 select * from table1 where status success);// 以上代码已经写完了下面是两个步骤分开的写法//TODO 3.transformation/查询// Table result tEnv.sqlQuery(select user_id,page_id,status from table1 where statussuccess);//输出到Kafka DDL// tEnv.executeSql(insert into table2 select * from result);//2. source-加载数据//3. transformation-数据处理转换//4. sink-数据输出//5. execute-执行// env.execute();}
}
