本文为学习笔记，会随着学习深入持续更新，仅供参考
1、什么是flink
Flink是一个批处理和流处理结合的统一计算框架和分布式处理引擎，可以实现数据的实时采集、实时计算和下游发送。
2、实现的原理和基本模块组成
flink实现一个简单程序的编程逻辑包括三部分：source、transform和sink。
编程模型.png
通常我们处理大数据的场景是在集群下完成的，flink的集群模型是这样工作的：
- 用户在提交编写好的 flink 工程时，会先创建一个client再进行提交，client将这个job提交给集群的JobManager；
- JobManager将job调度到一个TaskManager
- TaskManager实际负责执行计算的 Worker，在其上执行 flink job 的一组 task，一个taskManager就是一个jvm进程，可以用独立的线程来执行task，而一组task就包含：soruce、transform和sink；
- 执行完的task会将状态/结果返回client.
详细解释参考链接
集群的模型.png
3、常用API
flink建立初的目标是流批一体的处理引擎，它可以处理有界数据和无界数据。
有界数据集，就比如：文件、表、java集合；适用于DataSet api；
无界数据流，就比如：kafka持续存在的数据流；适用于DataStream api;
还有其他的如处理机器学习的Machine Learning api；
在上层处理flink提供了可供场景选择的api，内部的处理逻辑是一样的。
架构.png
常用的流式处理DataStream api的处理方法可以参考：Flink DataStream API 编程指南
Flink 提供了两种关系型 API 来做流批统一处理：Table API 和 SQL，无论是流数据还是历史数据，使用Table API和SQL都能够保证查询的一致性和准确性，更适合做数据的实时分析。
Table API 和 SQL直达链接
4、flink如何保持高效率，高可靠和状态一致性？常见问题有哪些
a)保持数据一致性
Checkpoint：Flink通过定期在作业的数据流中插入checkpoint，将作业的状态保存到持久化存储。这样，在发生故障时，作业可以从故障之前的状态恢复，避免了数据重复处理或丢失。
Exactly-Once Sink：Flink中的Sink操作符保证仅将一条记录写入外部系统，即使在发生故障时也不会重复写入相同的记录。这通过将Sink操作符的输出与应用的状态一起保存在持久化存储中实现。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 开启 checkpoint,每5000毫秒进行一次,如果作业失败，它会从最近的一次 checkpoint 恢复，保证数据一致性
env.enableCheckpointing(5000); env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
image.png
b)反压问题（消费速度小于生产速度）
反压的判断：TaskManager会定期进行反压检测，然后将结果返回给JobManager，JobManager在进行整体比例计算，结果反应在监控指标上
Flink metrics的官网文档链接
image.png
问题处理：
分析方式一.png
分析方式二.png
分析方式三.png
d) 数据倾斜（某个节点压力极大，导致任务失败）
产生原因：业务有严重的热点数据，错误使用keyBy/GroupBy，错误的使用了分组的key，认为生成热点数据；flink消费kafka数据时上下游并行度不一致（推荐：kafka的分区数和flink consumer的并行度一致）
解决：避免热点key的设计；自定义分区策略
避免热点key的设计解决思路.png
分区策略.png
e) 并行度设置
image.png
f) 如何做维表关联查询
场景：在处理流数据的时候需要用到数据库表的数据（如：将订单数据中省份id转化成省份数据等）
解决：1、小数据量的维表可以进行实时查询，但是需要考虑用线程池，妥善处理外部线程连接，及时关闭。
2、预加载维表全量数据。3、LRU缓存（冷热数据）
5、在springboot中如何实现数据转换集成
应用场景：接收kafka数据入库hdfs/doris，实时写入。
参考代码逻辑如下
<!--核心依赖-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>你的Flink版本</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>你的Flink版本</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>你的Flink版本</version>
</dependency>
<!--外部链接依赖-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-base</artifactId>
<version>你的Flink版本</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>你的Flink版本</version>
</dependency>
<!-- 如果需要写入Hadoop FileSystem，添加以下依赖 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-filesystem</artifactId>
<version>你的Flink版本</version>
</dependency>
<!-- 如果需要写入Hadoop HDFS，确保你的Hadoop版本兼容，并添加相应依赖 -->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>你的Hadoop版本</version>
</dependency>
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.IOException;
import java.util.Properties;
public class FlinkKafkaTLVToHadoop {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 配置Kafka消费者
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "kafka-broker:9092");
kafkaProps.setProperty("group.id", "my-group");
kafkaProps.setProperty("auto.offset.reset", "earliest");
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
"my-input-topic",
new SimpleStringSchema(),
kafkaProps
);
DataStream<String> kafkaStream = env.addSource(kafkaConsumer);
// 解析TLV数据
DataStream<MyParsedData> parsedDataStream = kafkaStream
.process(new TLVParserFunction()); // 实现自定义的解析函数
// 写入Hadoop HDFS
parsedDataStream.addSink(new HadoopFsSink()); // 实现自定义的Hadoop Sink
env.execute("Flink Kafka TLV to Hadoop Job");
}
// 自定义的Hadoop Fs Sink
public static class HadoopFsSink implements SinkFunction<MyParsedData> {
private FileSystem fileSystem;
@Override
public void invoke(MyParsedData value) throws IOException {
if (fileSystem == null) {
fileSystem = FileSystem.get(new java.net.URI("hdfs://namenode:8020"), new Configuration());
}
fileSystem.write(new Path("/path/to/hdfs/file"), value.toString());
}
}
// 自定义的解析函数
public static class TLVParserFunction extends RichMapFunction<String, MyParsedData> {
@Override
public MyParsedData map(String tlv) throws Exception {
// 实现具体的解析逻辑
return new MyParsedData(/* 解析后的数据 */);
}
}
// 自定义的数据类型，表示解析后的数据
public static class MyParsedData {
// 字段和构造函数
}
}
参考文件
1、Flink基本原理
2、官网文档
3、Flink学习教程-文档
4、Flink学习教程-视频
5、Flink常用API详解
