Flink常见面试题

server/2024/12/3 7:24:24/

1、Flink 的四大特征（基石）

2、Flink 中都有哪些 Source，哪些 Sink，哪些算子（方法）

预定义Source

基于本地集合的source（Collection-based-source）

基于文件的source（File-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-加载数据DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999);SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = dataStreamSource.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String s) throws Exception {String[] arr = s.split(",");return Tuple2.of(arr[0], Integer.valueOf(arr[1]));}});//3. transformation-数据处理转换SingleOutputStreamOperator<Tuple2<String, 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 status='success'");//输出到Kafka    DDL// tEnv.executeSql("insert into table2 select * from " + result);//2. source-加载数据//3. transformation-数据处理转换//4. sink-数据输出//5. execute-执行// env.execute();}
}


http://www.ppmy.cn/server/146956.html

相关文章

【Leetcode 每日一题】LCR 190. 加密运算

【Leetcode 每日一题】LCR 190. 加密运算

LCR 190. 加密运算 计算机安全专家正在开发一款高度安全的加密通信软件&#xff0c;需要在进行数据传输时对数据进行加密和解密操作。假定 dataA 和 dataB 分别为随机抽样的两次通信的数据量&#xff1a; 正数为发送量负数为接受量0 为数据遗失 请不使用四则运算符的情况下实…
阅读更多...
分布式资源调度——yarn 概述（资源调度基本架构和高可用的实现）

分布式资源调度——yarn 概述（资源调度基本架构和高可用的实现）

此文章是学习笔记&#xff0c;图片均来源于B站&#xff1a;哈喽鹏程 yarn详细介绍 1、yarn 简介1.1 yarn的简介1.2 yarn 的基本架构1.3. yarn 的高可用 2、yarn 调度策略、运维、监控2.1 yarn 的调度策略2.1.1 FIFO scheduler(先进先出)2.1.2 容量调度2.1.3 公平调度 2.2 yarn…
阅读更多...
计算机网络——数据链路层Mac帧详解

计算机网络——数据链路层Mac帧详解

目录 前言 一、以太网 二、Mac帧 三、MTU——最大传输单元 四、Mac帧的传输过程 1.ARP协议 2.RARP协议 前言 在之前&#xff0c;我们学习过网络层的IP协议&#xff0c;了解到IP协议解决了从哪里来&#xff0c;到哪里去的问题&#xff0c;也就是提供了将数据从A到B的能力…
阅读更多...
本地Docker部署个人在线音乐平台Melody结合内网穿透远程访问听音乐

本地Docker部署个人在线音乐平台Melody结合内网穿透远程访问听音乐

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…
阅读更多...
高职本科大数据实训平台及使用场景

高职本科大数据实训平台及使用场景

大数据实验实训平台是在云计算、大数据、人工智能等基础上&#xff0c;以智能化容器集群技术搭建的“一站式”大数据、人工智能等相关专业的实验、实训、实战平台&#xff0c;支撑“教—学—练—赛—研”等全闭环的专业建设、人才培养、实验室建设等整体解决方案。 大数据…
阅读更多...
深入学习MapReduce：原理解析与基础实战

深入学习MapReduce：原理解析与基础实战

标题&#xff1a;深入学习MapReduce&#xff1a;原理解析与基础实战 MapReduce是一种分布式计算框架&#xff0c;用于大规模数据的处理和分析。作为Hadoop生态系统的核心组件&#xff0c;MapReduce凭借其简单的编程模型和强大的并行计算能力&#xff0c;广泛应用于大数据领域。…
阅读更多...
H3C OSPF实验

H3C OSPF实验

实验拓扑 实验需求 按照图示配置 IP 地址按照图示分区域配置 OSPF &#xff0c;实现全网互通为了路由结构稳定&#xff0c;要求路由器使用环回口作为 Router-id&#xff0c;ABR 的环回口宣告进骨干区域 实验解法 一、配置IP地址 [R1]int l0 [R1-LoopBack0]ip add 1.1.1.1 32 […
阅读更多...
Path类使用---python

Path类使用---python

一、初始化 1.1、路径拼接 操作符/ .Path对象 / Path对象 .Path对象 / 字符串 .字符串 / Path对象 joinpath .joinpath(*other)在当前Path路径连接多个字符串返回新路径对象 二、分解 parts属性&#xff0c;会返回目录各部分的元组 三、获取路径 str获取路径字符…
阅读更多...
最新文章

Copyright @ 2022~2023