说明
这里使用 maven-shade-plugin 打包插件
依赖冲突问题
打包时，不要包含 flink自带的jar包，否则会因为冲突导致报错
解决冲突
方式一
在 flink自带的jar，添加 <scope>provided</scope>，打包时，不会包含这些包
缺点：此时工程是不能运行的，运行报错找不到类，为了运行，还需要去掉该配置，反复修改很麻烦
方式二
在 <properties> 中配置 scope，下面依赖都使用该配置，这样改动小
<properties>
<!-- 打包时，使用下面配置 -->
<dependency.scope>provided</dependency.scope>
<!-- 工程运行时，使用下面配置 -->
<!-- <dependency.scope>compile</dependency.scope>-->
</properties>
依赖：
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
<!-- 使用 dependency.scope 的配置 -->
<scope>${dependency.scope}</scope>
</dependency>
</dependencies>
总结
选择第二种方式
配置
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<flink.version>1.12.0</flink.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<dependency.scope>provided</dependency.scope>
<!-- <dependency.scope>compile</dependency.scope>-->
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
<scope>${dependency.scope}</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>${flink.version}</version>
<scope>${dependency.scope}</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>${flink.version}</version>
<scope>${dependency.scope}</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.30</version>
<scope>${dependency.scope}</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.30</version>
<scope>${dependency.scope}</scope>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
<scope>${dependency.scope}</scope>
</dependency>
</dependencies>
注意
flink-runtime-web_2.12 要设置 <scope>provided</scope>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web_2.12</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
注意
org.apache.bahir 不要用，因为是 scala 2.11 版本，与 flink 中的scala 2.12 版不匹配
<dependency>
<groupId>org.apache.bahir</groupId>
<artifactId>flink-connector-redis_2.11</artifactId>
<version>1.0</version>
</dependency>
添加打包插件
<build>
<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>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass></mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
常见错误
可能会报下面错误：
org.apache.flink.client.program.ProgramInvocationException: The program's entry point class 'ods.DataCleaning' caused an exception during initialization: Invalid signature file digest for Manifest main attributes
at org.apache.flink.client.program.PackagedProgram.loadMainClass(PackagedProgram.java:497)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
Caused by: java.lang.SecurityException: Invalid signature file digest for Manifest main attributes
at sun.security.util.SignatureFileVerifier.processImpl(SignatureFileVerifier.java:330)
分析
因为依赖jar包中的 META-INF 中有多余的 .SF 文件与当前jar包冲突
解决
在 <version> 下增加下面代码：
<configuration>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
例子
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.2</version>
<configuration>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass></mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
完整依赖
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<flink.version>1.12.0</flink.version>
</properties>
<dependencies>
<!--java-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.30</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.30</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
<scope>provided</scope>
</dependency>
<!-- kafka Dependency -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-files</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web_2.12</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.47</version>
</dependency>
<!--<dependency>-->
<!-- <groupId>org.apache.bahir</groupId>-->
<!-- <artifactId>flink-connector-redis_2.11</artifactId>-->
<!-- <version>1.0</version>-->
<!--</dependency>-->
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
</dependencies>
<build>
<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>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass></mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
java 类
package map;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
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.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;
public class TestKafka {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 定义kafka相关配置项
Properties pro = new Properties();
pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
// 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
DataStreamSource<String> dss = env.addSource(fkc);
/* 匿名类实现
* 第一个泛型，表示原数据的类型
* 第二个泛型，表示转换后的数据类型
*/
SingleOutputStreamOperator<UserLog> soso = dss.map(new MapFunction<String,UserLog>(){
@Override
public UserLog map(String s) throws Exception {
ObjectMapper mapper = new ObjectMapper();
UserLog result = mapper.readValue(s, UserLog.class);
return result;
}
});
soso.print();
env.execute();
}
}
log4j.properties
在 resources 目录下创建 log4j.properties 文件，内容如下：
# DEBUG 级别，即：所有信息都会输出
log4j.rootLogger=INFO, stdout
# 配置 stdout，输出到控制台
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
注意：不加此文件会显示警告
测试
打包
双击 maven 的 package，打包，结果有2个 jar 文件，一个文件比较大，这个文件就是打包好的文件
启动zk
登录 hadoop1，执行下面命令：
/program/bin/zk.sh start
启动kafka
登录 hadoop1，执行下面命令：
/program/bin/kafka.sh start
启动hadoop
/program/bin/hadoop.sh start
上传
略
per-job cluster模式启动
/program/flink-1.12.5/bin/flink run -m yarn-cluster -ytm 1000 -yjm 1000 -ys 1 -c map.TestKafka /program/flink-1.0-SNAPSHOT.jar
看到下图表示启动成功：
创建生产者、指定主题
登录 hadoop1，执行下面命令：
/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092,hadoop2:9092,hadoop3:9092 --topic kafka2flink
然后输入：
{"name":"李雷","url":"/jianpan.html","time":"2020-09-01 10:00:00"}
{"name":"lucy","url":"/lv.html","time":"2020-09-01 12:00:00"}
{"name":"lucy","url":"/dior.html","time":"2020-09-01 15:00:00"}
执行结果
