引言
gitee地址：https://gitee.com/shawsongyue/aurora.git
源码直接下载可运行，模块：aurora_flink
Flink 版本：1.18.0
Jdk 版本：11
1.创建mavenx项目
2.包结构
3.引入pom依赖
tips：transformer处写主启动类
<?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"> <modelVersion>4.0.0</modelVersion> <groupId>com.xsy</groupId> <artifactId>aurora_flink</artifactId> <version>1.0-SNAPSHOT</version> <!--属性设置--> <properties> <!--java_JDK版本--> <java.version>11</java.version> <!--maven打包插件--> <maven.plugin.version>3.8.1</maven.plugin.version> <!--编译编码UTF-8--> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <!--输出报告编码UTF-8--> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <!--json数据格式处理工具--> <fastjson.version>1.2.75</fastjson.version> <!--log4j版本--> <log4j.version>2.17.1</log4j.version> <!--flink版本--> <flink.version>1.18.0</flink.version> <!--scala版本--> <scala.binary.version>2.11</scala.binary.version> <!--log4j依赖--> <log4j.version>2.17.1</log4j.version> </properties> <!--通用依赖--> <dependencies> <!-- json --> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>${fastjson.version}</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> </dependency> <!--================================集成外部依赖==========================================--> <!--集成日志框架 start--> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-slf4j-impl</artifactId> <version>${log4j.version}</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-api</artifactId> <version>${log4j.version}</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>${log4j.version}</version> </dependency> <!--集成日志框架 end--> </dependencies> <!--编译打包--> <build> <finalName>${project.name}</finalName> <!--资源文件打包--> <resources> <resource> <directory>src/main/resources</directory> </resource> <resource> <directory>src/main/java</directory> <includes> <include>**/*.xml</include> </includes> </resource> </resources> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.1.1</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <artifactSet> <excludes> <exclude>org.apache.flink:force-shading</exclude> <exclude>org.google.code.flindbugs:jar305</exclude> <exclude>org.slf4j:*</exclude> <excluder>org.apache.logging.log4j:*</excluder> </excludes> </artifactSet> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass>org.xsy.sevenhee.flink.TestStreamJob</mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> <!--插件统一管理--> <pluginManagement> <plugins> <!--maven打包插件--> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> <version>${spring.boot.version}</version> <configuration> <fork>true</fork> <finalName>${project.build.finalName}</finalName> </configuration> <executions> <execution> <goals> <goal>repackage</goal> </goals> </execution> </executions> </plugin> <!--编译打包插件--> <plugin> <artifactId>maven-compiler-plugin</artifactId> <version>${maven.plugin.version}</version> <configuration> <source>${java.version}</source> <target>${java.version}</target> <encoding>UTF-8</encoding> <compilerArgs> <arg>-parameters</arg> </compilerArgs> </configuration> </plugin> </plugins> </pluginManagement> </build> <!--配置Maven项目中需要使用的远程仓库--> <repositories> <repository> <id>aliyun-repos</id> <url>https://maven.aliyun.com/nexus/content/groups/public/</url> <snapshots> <enabled>false</enabled> </snapshots> </repository> </repositories> <!--用来配置maven插件的远程仓库--> <pluginRepositories> <pluginRepository> <id>aliyun-plugin</id> <url>https://maven.aliyun.com/nexus/content/groups/public/</url> <snapshots> <enabled>false</enabled> </snapshots> </pluginRepository> </pluginRepositories> </project>
4.增加log4j2.properties配置
tips：resource目录下增加该配置，主要用于日志打印
rootLogger.level=INFO rootLogger.appenderRef.console.ref=ConsoleAppender appender.console.name=ConsoleAppender appender.console.type=CONSOLE appender.console.layout.type=PatternLayout appender.console.layout.pattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n log.file=D:\\tmp
5.创建主启动类
tips：编写了一个简单的有界数据流处理demo程序
- step1：创建flink程序运行所需环境
- step2：创建数据集
- step3：把有限数据集转换为数据源
- step4：简单通过flatmap处理数据
- step5：输出最终结果
- step6：启动任务
package com.aurora; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.SinkFunction; import org.apache.flink.util.Collector; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Random; import java.util.UUID; /** * @author 浅夏的猫 * @description 主启动类 * @date 22:46 2024/1/13 */ public class Application { private static final Logger logger = LoggerFactory.getLogger(Application.class); public static void main(String[] args) throws Exception { //1.创建flink程序运行所需环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //2.创建数据集 ArrayList<String> list = new ArrayList<>(); list.add("001"); list.add("002"); list.add("003"); //3.把有限数据集转换为数据源 DataStreamSource<String> dataStreamSource = env.fromCollection(list).setParallelism(1); //4.简单通过flatmap处理数据， SingleOutputStreamOperator<String> flatMap = dataStreamSource.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String record, Collector<String> collector) throws Exception { //数据追加随机数 String uuidRecord=record+ UUID.randomUUID().toString(); //当前环节处理完需要传递数据给下个环节 collector.collect(uuidRecord); } }); //5.输出最终结果 flatMap.addSink(new SinkFunction<String>() { @Override public void invoke(String value) throws Exception { logger.info("当前正在处理的数据:{}",value); } }).setParallelism(1); //6.启动任务 env.execute(); } }
6.构建打jar包
7.flinkUI页面部署
1.点击add new上传对应的应用包
2.主类填写com.aurora.Application
3.检查任务running状态，大概几秒钟跑完
