快捷導(dǎo)航

SpringBoot集成flink全過程

更新時(shí)間：2025年01月14日 14:45:36 作者：A塵埃

文章介紹了Flink作為批處理和流處理結(jié)合的統(tǒng)一計(jì)算框架,特別是其強(qiáng)大的流處理能力,文章還詳細(xì)描述了如何在本地和集群環(huán)境中搭建Flink,并通過Netcat工具生成一個(gè)無界流測試,文章最后提供了啟動(dòng)socket流的步驟和示例代碼,希望對讀者有所幫助

SpringBoot集成flink

Flink是一個(gè)批處理和流處理結(jié)合的統(tǒng)一計(jì)算框架，其核心是一個(gè)提供了數(shù)據(jù)分發(fā)以及并行化計(jì)算的流數(shù)據(jù)處理引擎。

最大亮點(diǎn)是流處理，最適合的應(yīng)用場景是低時(shí)延的數(shù)據(jù)處理。

場景

高并發(fā)pipeline處理數(shù)據(jù)，時(shí)延毫秒級，且兼具可靠性。

環(huán)境搭建

①、安裝flink

https://nightlies.apache.org/flink/flink-docs-master/zh/docs/try-flink/local_installation/

②、安裝Netcat

Netcat（又稱為NC）是一個(gè)計(jì)算機(jī)網(wǎng)絡(luò)工具，它可以在兩臺(tái)計(jì)算機(jī)之間建立 TCP/IP 或 UDP 連接。

用于測試網(wǎng)絡(luò)中的端口，發(fā)送文件等操作。

進(jìn)行網(wǎng)絡(luò)調(diào)試和探測，也可以進(jìn)行加密連接和遠(yuǎn)程管理等高級網(wǎng)絡(luò)操作

yum install -y nc # 安裝nc命令
nc -lk 8888 # 啟動(dòng)socket端口

無界流之讀取socket文本流

一、依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>springboot-demo</artifactId>
        <groupId>com.et</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>flink</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-autoconfigure</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- 添加 Flink 依賴 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>1.17.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.17.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>1.17.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-base -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</artifactId>
            <version>1.17.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-files</artifactId>
            <version>1.17.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>1.17.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web</artifactId>
            <version>1.17.0</version>
        </dependency>


    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>META-INF/spring.handlers</resource>
                                </transformer>
                                <transformer
                                        implementation="org.springframework.boot.maven.PropertiesMergingResourceTransformer">
                                    <resource>META-INF/spring.factories</resource>
                                </transformer>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>META-INF/spring.schemas</resource>
                                </transformer>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.et.flink.job.SocketJob</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

二、SoketJob

public class SocketJob{
	
	public static void main(String[] args)throws Exception{
		
		// 創(chuàng)建執(zhí)行環(huán)境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 指定并行度,默認(rèn)電腦線程數(shù)
        env.setParallelism(3);
        // 讀取數(shù)據(jù)socket文本流 指定監(jiān)聽 IP 端口 只有在接收到數(shù)據(jù)才會(huì)執(zhí)行任務(wù)
        DataStreamSource<String> socketDS = env.socketTextStream("172.24.4.193", 8888);

        // 處理數(shù)據(jù): 切換、轉(zhuǎn)換、分組、聚合 得到統(tǒng)計(jì)結(jié)果
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
                .flatMap(
                        (String value, Collector<Tuple2<String, Integer>> out) -> {
                            String[] words = value.split(" ");
                            for (String word : words) {
                                out.collect(Tuple2.of(word, 1));
                            }
                        }
                )
                .setParallelism(2)
                // // 顯式地提供類型信息:對于flatMap傳入Lambda表達(dá)式，系統(tǒng)只能推斷出返回的是Tuple2類型，而無法得到Tuple2<String, Long>。只有顯式設(shè)置系統(tǒng)當(dāng)前返回類型，才能正確解析出完整數(shù)據(jù)
                .returns(new TypeHint<Tuple2<String, Integer>>() {
                })
//                .returns(Types.TUPLE(Types.STRING,Types.INT))
                .keyBy(value -> value.f0)
                .sum(1);


        // 輸出
        sum.print();

        // 執(zhí)行
        env.execute();
	}
}

測試：

啟動(dòng)socket流：