一次记录 flink job 消费kafka 迁移pulsar踩坑过程

背景简述

业务上,原有的kafka集群迁移pulsar 后续会下线kafak集群,原有的一些消费kafka topic 的任务和进程需要迁移至pulsar 并下线旧的消费kafka任务。目前在迁移期间,上报的消息会双写到kafka pulsar,消费组的offset二者是独立的。

待迁移的flink job 之前flink 版本是 1.9.1(scala 2.12) 消费kafak

FlinkKafkaConsumer<String> myConsumer = KafkaConsumerFactory.get(topic, group);
env.addSource(myConsumer )...

这次迁移 job的版本集群环境需要 升级为 1.13(scala 2.12) 消费pulsar

/*job main 方法 消费topic 迁移到pulsar*/
FlinkPulsarSource<String> myConsumer = PulsarSourceFactory.getSource(topic, group);
env.addSource(myConsumer)...

/**
 * @author: xiejiahao
 * Date: 2021/12/22 11:28
 * Description:
 */
public class PulsarSourceFactory {
    public static final String serviceUrl =
            "pulsar://pulsar-***:6651";
    public static final String adminUrl =
            "http://pulsar-*****:8441";

    public static FlinkPulsarSource<String> getSource(String topic, String group) {

        FlinkPulsarSource<String> flinkPulsarSource =
                new FlinkPulsarSource<>(
                        serviceUrl,
                        adminUrl,
                        PulsarDeserializationSchema.valueOnly(new SimpleStringSchema()),
                        buildProperties(topic, group)).setStartFromSubscription(group);

        return flinkPulsarSource;
    }

    private static Properties buildProperties(String topic, String group) {
        Properties props = new Properties();
        props.put("topic", "persistent://****/" + topic);
        props.put("pulsar.reader.subscriptionRolePrefix", group);
        props.setProperty("auth-plugin-classname", "org.apache.pulsar.client.impl.auth.AuthenticationToken");
        props.setProperty("auth-params", "token:" + AppConfig.get("pulsar_topic_token"));

        // earliest
        // 当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
        // latest
        // 当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
        // none
        // topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
        String offset = AppConfig.get("kafka.offset", "");
        if (!StringUtils.isBlank(offset)) {
            props.put("auto.offset.reset", offset);
        }
        // props.put("client.id", "h_c_" + ThreadLocalRandom.current().nextLong());
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", ByteArrayDeserializer.class.getName());
        return props;
    }
}

开发过程需要注意的点(踩坑记录)

flink支持消费pulsar 必备的依赖

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-connector-pulsar_2.12</artifactId>
   <version>1.13.0</version>
   <scope>${scope.version}</scope>
</dependency>
<dependency>
   <groupId>org.apache.pulsar</groupId>
   <artifactId>pulsar-client-all</artifactId>
   <version>2.8.0</version>
   <scope>${scope.version}</scope>
</dependency>

pom文件里面依赖配置的 ${scope.version} 本地测试时设置为compile 打包提交到集群时 为避免和集群依赖冲突 改为provided 然后再打包上传

上传jar包时,应确认jar包中不存在与flink, pulsar相关的依赖包,避免运行作业时与平台自带的依赖包发生冲突。

当在本地进行测试时,因为job之前的版本是1.9.1 本地测试时需要/org/apache/flink/flink-core/1.13.0/flink-core-1.13.0.jar下载依赖包到本地并进行替换。

依赖冲突解决

这里 我的flink job中需要使用到consul 用来服务发现 codis集群,但是consul内置的okhttp okio版本都要高于flink集群上的版本,因为版本冲突 运行Job时直接报错 NoSuchMethodError

这类问题的解决办法是 使用maven-shade-plugin 对版本冲突的报的类名直接重命名 然后才重新编译

<plugin>
   <groupId>org.apache.maven.plugins</groupId>
   <artifactId>maven-shade-plugin</artifactId>
   <version>3.1.1</version>
   <executions>
      <execution>
         <phase>package</phase>
         <goals>
            <!--使用shade 解决本地和集群包版本 依赖冲突-->
            <goal>shade</goal>
         </goals>
         <configuration>
            <transformers>
               <transformer
                  implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                  <mainClass>com.test.App</mainClass>
               </transformer>
            </transformers>

            <relocations>            
               <relocation>
                  <pattern>okio</pattern>
                  <shadedPattern>okio.ad.shade</shadedPattern>
               </relocation>

               <relocation>
                  <pattern>okhttp3</pattern>
                  <shadedPattern>okhttp3.ad.shade</shadedPattern>
               </relocation>
            </relocations>
         </configuration>
      </execution>
   </executions>
</plugin>

在重新打包后, 通过jd-gui 查看jar 可发现pom 依赖的 okio okhttp3 已经重命名,这样可解决job 自身依赖的jar 与平台不兼容问题
一次记录 flink job 消费kafka 迁移pulsar踩坑过程

日志框架冲突导致job 运行时无业务日志输出

我的flink job自身使用的日志框架是 common-log, 但是flink集群默认使用的日志框架是log4j 二者发生冲突导致 job 自身的业务日志无法输出, 这里改为slf4j + log4j2

具体操作是

<dependency>
   <groupId>org.slf4j</groupId>
   <artifactId>slf4j-api</artifactId>
   <version>1.7.25</version>
   <scope>${scope.version}</scope>
</dependency>

<dependency>
   <groupId>org.slf4j</groupId>
   <artifactId>slf4j-log4j12</artifactId>
   <version>1.7.21</version>
   <scope>${scope.version}</scope>
</dependency>

<dependency>
   <groupId>log4j</groupId>
   <artifactId>log4j</artifactId>
   <version>1.2.17</version>
   <scope>${scope.version}</scope>
</dependency>

并配置log4j.properties

log4j.rootLogger=info,console  
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.Threshold=INFO
log4j.appender.console.ImmediateFlush=true
log4j.appender.console.Target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=[%-5p] %d(%r) --> [%t] %l: %m %x %n

这里具体的配置 参考的是:https://www.shuzhiduo.com/A/gAJGqkEodZ/

值得注意的是 log4j 已经报出有安全漏洞问题,当你的项目pom文件中 配置log4j依赖 打包到平台一定要确定scop为provided 以便使用集群上统一的高版本的log4j。

timewindow算子过期不能使用导致sink输出没有数据

这个坑 应该是本次迁移 折腾最久的一个坑了,因为之前日志框架冲突的问题没有解决 sink输出没有数据 也无法定位到是哪一块出现问题。

job 自身逻辑中 对数据的唯一id hashcode keyBy分区之后 timeWindow(60s) 将批量的数据 写入db中进行存储,但是日志观察window函数 一直没有正常日志输出,sink输出的db 也一直没有数据。

这里对flink 环境设置

if (arg.contains("TEST")){
    env.disableOperatorChaining();
}

重新提交 观察数据在各个算子建的流动情况时发现 window算子这里只有记录接收 无记录发送到下游
一次记录 flink job 消费kafka 迁移pulsar踩坑过程

初步怀疑是历史代码的timewindow算子 在flink 1.13的版本里面有bug 导致window执行出现异常

解决办法:

这里测试 将keyBy+window 简化成一个flatmap 算子有正常输入输出了
一次记录 flink job 消费kafka 迁移pulsar踩坑过程
一次记录 flink job 消费kafka 迁移pulsar踩坑过程

考虑到业务场景 最好是批量写入tidb 减少tidb的连接数, 这里后面改用了countwindow来做window函数 默认的size是100

int tidbCountWindowsSize = AppConfig.get("main.tidb.countWindowsSize", 100);
int tidbCountTriggerTimeout = AppConfig.get("main.tidb.CountTriggerTimeout", 1000);
LogCountWithTimeoutTrigger tidbCountWithTimeoutTrigger = LogCountWithTimeoutTrigger.of(tidbCountWindowsSize,
        tidbCountTriggerTimeout);

DataStream<DBRecord> dbList = groupkeyStream
        .countWindow(tidbCountWindowsSize).trigger(tidbCountWithTimeoutTrigger)
        .process(new OrderCountWindowFunction(tidbCountWindowsSize)).name("order_countWindow_records");

部署后 数据在各个算子的流动情况一切正常
一次记录 flink job 消费kafka 迁移pulsar踩坑过程
对比测试环境tidb和线上消费kafka sink输出到tidb 的数据情况,因为测试环境使用的countwindow(默认是100) 数据从redis获取去重的数据更新到tidb 更快,性能要优于之前逻辑。

上一篇:Flink流计算学习 一


下一篇:Ubuntu18.04使用nginx配置反向代理-(3)