八、spark streaming 如何保证消费EOS

阳光穿透心脏的1/2处 2023-06-02 07:55 10阅读 0赞

前段时间，一直有人问 spark streaming 偏移量问题。一个正常或者线上的sparkstreaming执行顺序：

1.根据group获取kafka当前消费的位置

2.创建Dstream

3.rdd算子或者逻辑

4.将计算结果保存到数据库中

5.提交偏移量

正常情况下，spark streaming消费是没办法实现EOS。

### **一、为什么没办法实现EOS？** ###

不管你先执行第4步或者先执行第5步，都是有问题的。先执行第4步，那么执行完第4步，程序挂了，偏移量没提交，下次程序重新启动的时候，消息又要重新消费，这是 at leastonce，至少一次消费，所以大部分情况下很多公司都是这样的，当然可以通过数据库来保证结果的幂等性（比如hbase，不管插入多少条，最终结果都是一条，因为rowkey相同）。第5步先执行就是扯淡了。。  
那怎么才能保证消费的EOS呢？

答：事务！

将第4部和第5部处在同一个事务中就能保证EOS。如果将第4部保存在redis中，第5部偏移量保存到zk上或者hbase上这也是不行的，得自己实现分布式事务。简单点，**就是让第4步和第5步保存在同一个数据库中，然后这个数据库带事务功能就ok。**

### 二、下面用mysql举个例子 ###

1、首先创建一张偏移量表：

CREATE TABLE `offset` (
      `id` varchar(255) NOT NULL,
      `topic` varchar(255) DEFAULT NULL,
      `partition` int(11) DEFAULT NULL,
      `offsets` bigint(255) DEFAULT NULL,
      `group` varchar(255) DEFAULT NULL,
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

2、定义一个抽象偏移量工具类

**
     * Created by zhuenjun on 2019/3/24.
    为什么需要这个类?像我公司目前是把偏移量提交到原生的kafka内部，kafka09版本是提交到zk上，
    自从上了kafka2.0.1，kafka自己都不把偏移量保存到zk上了，那我还保存到zk上干嘛。。。
    这个类的作用就是根据topic，group获取外置的偏移量。
     */
    public abstract class AbstractKafkaOffsetsTool implements Closeable {
        final Logger logger = LoggerFactory.getLogger(this.getClass());
        private static final String CLIENT_ID = "client-offsets";
        String brokerServers;
        public String topic;
        public String group;
        private KafkaConsumer<byte[], byte[]> kafkaConsumer;
    
        public AbstractKafkaOffsetsTool(String brokerServers, String topic, String group) {
            this.brokerServers = brokerServers;
            this.topic = topic;
            this.group = group;
        }
    
        /**
         * 初始化的消费偏移量(不能直接拿来用)
         *
         * @return
         */
        protected abstract Map<TopicPartition, Long> initConsumserOffsets(String topic, String group);

3.获取mysql中的消息偏移量

class MysqlOffsets extends SparkStreamingOffsets {
      private static final String DRIVERCLASS = "com.mysql.jdbc.Driver";
      private static final String URL = 
            "jdbc:mysql://localhost:3306?serverTimezone=UTC&characterEncoding=utf-8";
      private static final String USERNAME = "root";
      private static final String PASSWORD = "root";
      JdbcPool jdbcPool = new JdbcPool(
              DRIVERCLASS,
              URL,
              USERNAME,
              PASSWORD);
    
      public MysqlOffsets(String group) {
          super(group);
      }
    
      @Override
      public Map<TopicPartition, Long> getConsumserOffsets(String brokerServer, String topic) {
          AbstractKafkaOffsetsTool kafkOffsetsTool = new AbstractKafkaOffsetsTool(brokerServer, topic, group) {
              private static final String sql = "select topic,`partition`,`offsets` from 
                                TEST.`OFFSET` where topic=? and `group`=?";
    
              @Override
              protected Map<TopicPartition, Long> initConsumserOffsets(String topic, String group) {
                  Map<TopicPartition, Long> topicPartitionLongMap = new HashMap<>();
                  jdbcPool.query(sql, new Object[]{topic, group}, result -> {
                      while (result.next()) {
                          topicPartitionLongMap.put(
                                  new TopicPartition(
                                          result.getString("topic"),
                                          result.getInt("partition")
                                  ),
                                  result.getLong("offsets"));
                      }
                  });
                  return topicPartitionLongMap;
              }
          };
          Map<TopicPartition, Long> consumserOffsets = kafkOffsetsTool.getConsumserOffsets();
          return consumserOffsets;
      }

**有人会在代码中发现，直接从mysql查出来就好，为何还要kafkOffsetsTool.getConsumserOffsets()这个方法？**

直接从mysql拿出来的偏移量是不能直接用的？为何？你的topic有可能会被删除（或者删除后又创建），但是你的偏移量保存的地方却在mysql，这样获取到的偏移量消费不到消息可不行。所以这里还有个逻辑就是你**获取到的偏移量得和kafka内部的topic的偏移量有个对比**，这步很重要，不然会有很多坑的！！！

@Override
       public void commitOffsets(OffsetRange[] offsetRangesResult, JavaInputDStream<ConsumerRecord<String, String>> directStream) throws Exception {
           String upsertOffsetsSql = "INSERT INTO TEST.`OFFSET` (id,topic,`partition`,`offsets`,`group`)\n" +
                   "VALUES(?,?,?,?,?) ON DUPLICATE KEY UPDATE offsets = ?";
           Connection connection = jdbcPool.getConnection();
           //设置false就是开启事务
           connection.setAutoCommit(false);
           try {
               //提交偏移量
               JdbcUtils.executePre(connection, upsertOffsetsSql, new ArrayList() {
       {
                   for (OffsetRange offsetRange : offsetRangesResult) {
                       //id="group_topic_partition"
                       add(new Object[]{
                               StringUtils.join(new Object[]{group, offsetRange.topic(), offsetRange.partition()}, "_"),
                               offsetRange.topic(),
                               offsetRange.partition(),
                               offsetRange.untilOffset(),
                               group,
                               offsetRange.untilOffset()
                       });
                   }
               }});
               //修改结果
               JdbcUtils.execute(connection, "INSERT INTO TEST.test(id,name) value('1','spark')");
               //提交事务
               connection.commit();
           } catch (Exception e) {
               //回滚
               connection.rollback();
           }
       }

昨天还有个小伙伴问我，如果偏移量提交到mysql，数据不是会膨胀的很厉害。。mysql也是可以实现upsert（replace into） 的方式的，通过ON DUPLICATE KEY UPDATE。上面还有一个spark的细节，mysql偏移量的提交是在driver端，而计算结果通常在executor，所以这块有个注意的地方就是需要把executor的计算结果汇集到driver端更新，我这里是方面写代码，就偷懒了写死了持久化逻辑，也没抽离啥的。。

最后一句话，剩余代码自己脑补！  
拒绝伸手党！！！！！！  
拒绝伸手党！！！！！！  
拒绝伸手党！！！！！！