本文為學(xué)習(xí)筆記磕瓷，會(huì)隨著學(xué)習(xí)深入持續(xù)更新膘螟，僅供參考

1笙僚、什么是flink
Flink是一個(gè)批處理和流處理結(jié)合的統(tǒng)一計(jì)算框架和分布式處理引擎芳肌，可以實(shí)現(xiàn)數(shù)據(jù)的實(shí)時(shí)采集、實(shí)時(shí)計(jì)算和下游發(fā)送味咳。

2庇勃、實(shí)現(xiàn)的原理和基本模塊組成
flink實(shí)現(xiàn)一個(gè)簡(jiǎn)單程序的編程邏輯包括三部分：source、transform和sink槽驶。

編程模型.png

1. 用戶在提交編寫(xiě)好的 flink 工程時(shí)，會(huì)先創(chuàng)建一個(gè)client再進(jìn)行提交掂铐，client將這個(gè)job提交給集群的JobManager罕拂；
2. JobManager將job調(diào)度到一個(gè)TaskManager
3. TaskManager實(shí)際負(fù)責(zé)執(zhí)行計(jì)算的 Worker，在其上執(zhí)行 flink job 的一組 task全陨，一個(gè)taskManager就是一個(gè)jvm進(jìn)程爆班，可以用獨(dú)立的線程來(lái)執(zhí)行task，而一組task就包含：soruce辱姨、transform和sink柿菩；
4. 執(zhí)行完的task會(huì)將狀態(tài)/結(jié)果返回client.
   詳細(xì)解釋參考鏈接
   
   集群的模型.png

3、常用API
flink建立初的目標(biāo)是流批一體的處理引擎雨涛，它可以處理有界數(shù)據(jù)和無(wú)界數(shù)據(jù)枢舶。
有界數(shù)據(jù)集懦胞，就比如：文件、表凉泄、java集合躏尉；適用于DataSet api；
無(wú)界數(shù)據(jù)流后众，就比如：kafka持續(xù)存在的數(shù)據(jù)流胀糜；適用于DataStream api;
還有其他的如處理機(jī)器學(xué)習(xí)的Machine Learning api；
在上層處理flink提供了可供場(chǎng)景選擇的api，內(nèi)部的處理邏輯是一樣的。

架構(gòu).png

4、flink如何保持高效率，高可靠和狀態(tài)一致性痊臭？常見(jiàn)問(wèn)題有哪些
a)保持?jǐn)?shù)據(jù)一致性
Checkpoint：Flink通過(guò)定期在作業(yè)的數(shù)據(jù)流中插入checkpoint，將作業(yè)的狀態(tài)保存到持久化存儲(chǔ)登夫。這樣广匙，在發(fā)生故障時(shí)，作業(yè)可以從故障之前的狀態(tài)恢復(fù)恼策，避免了數(shù)據(jù)重復(fù)處理或丟失鸦致。
Exactly-Once Sink：Flink中的Sink操作符保證僅將一條記錄寫(xiě)入外部系統(tǒng)，即使在發(fā)生故障時(shí)也不會(huì)重復(fù)寫(xiě)入相同的記錄涣楷。這通過(guò)將Sink操作符的輸出與應(yīng)用的狀態(tài)一起保存在持久化存儲(chǔ)中實(shí)現(xiàn)分唾。

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 開(kāi)啟 checkpoint,每5000毫秒進(jìn)行一次,如果作業(yè)失敗，它會(huì)從最近的一次 checkpoint 恢復(fù)狮斗，保證數(shù)據(jù)一致性
        env.enableCheckpointing(5000); env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

image.png

image.png

分析方式一.png

分析方式二.png

分析方式三.png

避免熱點(diǎn)key的設(shè)計(jì)解決思路.png

分區(qū)策略.png

image.png

5猎荠、在springboot中如何實(shí)現(xiàn)數(shù)據(jù)轉(zhuǎn)換集成
應(yīng)用場(chǎng)景：接收kafka數(shù)據(jù)入庫(kù)hdfs/doris，實(shí)時(shí)寫(xiě)入蜀备。
參考代碼邏輯如下

 <!--核心依賴-->
   <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>
    <!-- 如果需要寫(xiě)入Hadoop FileSystem关摇，添加以下依賴 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-filesystem</artifactId>
        <version>你的Flink版本</version>
    </dependency>
    <!-- 如果需要寫(xiě)入Hadoop HDFS，確保你的Hadoop版本兼容碾阁，并添加相應(yīng)依賴 -->
    <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消費(fèi)者
        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數(shù)據(jù)
        DataStream<MyParsedData> parsedDataStream = kafkaStream
            .process(new TLVParserFunction()); // 實(shí)現(xiàn)自定義的解析函數(shù)
 
        // 寫(xiě)入Hadoop HDFS
        parsedDataStream.addSink(new HadoopFsSink()); // 實(shí)現(xiàn)自定義的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());
        }
    }
 
    // 自定義的解析函數(shù)
    public static class TLVParserFunction extends RichMapFunction<String, MyParsedData> {
        @Override
        public MyParsedData map(String tlv) throws Exception {
            // 實(shí)現(xiàn)具體的解析邏輯
            return new MyParsedData(/* 解析后的數(shù)據(jù) */);
        }
    }
 
    // 自定義的數(shù)據(jù)類型输虱，表示解析后的數(shù)據(jù)
    public static class MyParsedData {
        // 字段和構(gòu)造函數(shù)
    }
}

參考文件
1、Flink基本原理
2脂凶、官網(wǎng)文檔
3宪睹、Flink學(xué)習(xí)教程-文檔
4、Flink學(xué)習(xí)教程-視頻
5蚕钦、Flink常用API詳解