温故知新：初识flink

本文为学习笔记，会随着学习深入持续更新，仅供参考

1、什么是flink
Flink是一个批处理和流处理结合的统一计算框架和分布式处理引擎，可以实现数据的实时采集、实时计算和下游发送。

---

2、实现的原理和基本模块组成
flink实现一个简单程序的编程逻辑包括三部分：source、transform和sink。

编程模型.png
通常我们处理大数据的场景是在集群下完成的，flink的集群模型是这样工作的：

1. 用户在提交编写好的 flink 工程时，会先创建一个client再进行提交，client将这个job提交给集群的JobManager；
2. JobManager将job调度到一个TaskManager
3. TaskManager实际负责执行计算的 Worker，在其上执行 flink job 的一组 task，一个taskManager就是一个jvm进程，可以用独立的线程来执行task，而一组task就包含：soruce、transform和sink；
4. 执行完的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详解