前段时间，一直有人问 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的方式的，通过ON DUPLICATE KEY UPDATE。上面还有一个spark的细节，mysql偏移量的提交是在driver端，而计算结果通常在executor，所以这块有个注意的地方就是需要把executor的计算结果汇集到driver端更新，我这里是方面写代码，就偷懒了写死了持久化逻辑，也没抽离啥的。。
最后一句话，剩余代码自己脑补！
拒绝伸手党！！！！！！
拒绝伸手党！！！！！！
拒绝伸手党！！！！！！
重要的事情说3遍！！！当然有什么不懂，可以互相交流。微信号 cto_zej，加我备注 大数据。
