如何获取流式应用程序中checkpoint的最新offset
对于流式应用程序,保证应用7*24小时的稳定运行,是非常必要的。因此对于计算引擎,要求必须能够适应与应用程序逻辑本身无关的问题(比如driver应用失败重启、网络问题、服务器问题、JVM崩溃等),具有自动容错恢复的功能。
目前,Spark(Spark Streaming/Structured Streaming)和Flink的checkpoint机制,就是处理类似情况,实现容错机制的核心利器。
对于Flink:
为了保证其高可用、Exactly Once的特性,提供了一套强大的checkpoint机制,它能够根据配置周期性地基于流中各个operator的状态来生成快照,从而将这些状态数据定期持久化存储下来,当Flink程序一旦出现故障时,能够将整个应用流程序恢复到故障前的某一种态,从而修正因为故障带来的程序数据状态中断。
对于Spark:
在流式应用中,Spark Streaming/Structured Streaming会将关于应用足够多的信息checkpoint到高可用、高容错的分布式存储系统,如HDFS中,以便从故障中进行恢复。checkpoint有两种类型的数据:
- 数据checkpoint
对于一些复杂程序,比如跨多个批次组合数据的有状态转换,生成的RDD依赖于先前批次的RDD,导致依赖链的长度随批次的增加而增加。因为故障恢复时间与依赖链成正比,从而导致恢复时间也跟着增长。因此就有必要周期性的将RDD checkpoint到可靠的分布式存储系统中,以此切断依赖链。
这在Spark中的状态算子,如mapWithState、updateStateByKey中尤为常见。
- 元数据checkpoint
顾名思义,就是将定义流式应用程序中的信息保存到容错系统中,用于从运行流应用程序的driver节点发生故障时,进行容错恢复。元数据包括:
- 配置:用于创建流应用程序DStream操作:
- 定义流应用程序的DStream操作集
- 未完成的批次:未完成的批次job
本文的重点不在于checkpoint具体含义,而是以Spark为例,阐述如何通过程序获取checkpoint中最新的offset,以此为思路,来解决生产中的实际问题。
通常我们会checkpoint到HDFS,首先来看一下checkpoint信息:
offsets目录记录了每个批次中的offset,此目录中的第N条记录表示当前正在处理,第N-1个及之前的记录指示哪些偏移已处理完成。
/bigdatalearnshare/checkpointLocation/binlog-2-kafka/commits
/bigdatalearnshare/checkpointLocation/binlog-2-kafka/metadata
/bigdatalearnshare/checkpointLocation/binlog-2-kafka/offsets
/bigdatalearnshare/checkpointLocation/binlog-2-kafka/receivedData
/bigdatalearnshare/checkpointLocation/binlog-2-kafka/sources
hdfs dfs -ls /bigdatalearnshare/checkpointLocation/binlog-2-kafka/offsets
/bigdatalearnshare/checkpointLocation/binlog-2-kafka/offsets/0
/bigdatalearnshare/checkpointLocation/binlog-2-kafka/offsets/1
/bigdatalearnshare/checkpointLocation/binlog-2-kafka/offsets/2
hdfs dfs -cat /bigdatalearnshare/checkpointLocation/binlog-2-kafka/offsets/2
v1
{"batchWatermarkMs":0,"batchTimestampMs":1590632490083,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.multipleWatermarkPolicy":"min","spark.sql.streaming.aggregation.stateFormatVersion":"2","spark.sql.shuffle.partitions":"1"}}
2400000001667289
最终获取最新offset的程序示例:
/**
* @Author bigdatalearnshare
*/
object ReadOffsets {
def main(args: Array[String]): Unit = {
val path = "/bigdatalearnshare/checkpointLocation/binlog-2-kafka/offsets/2"
val fs = FileSystem.get(new Configuration())
val lastFile = fs.listStatus(new Path(path)).filterNot(_.getPath.getName.endsWith(".tmp.crc"))
.map { fileName =>
(fileName.getPath.getName.split("/").last.toInt, fileName.getPath)
}.maxBy(_._1)._2
val offset = readFile(lastFile.toString).split("n").last
assert("2400000001667289".equals(offset))
}
def readFile(path: String): String = {
val fs = FileSystem.get(new Configuration())
var br: BufferedReader = null
var line: String = null
val result = ArrayBuffer.empty[String]
try {
br = new BufferedReader(new InputStreamReader(fs.open(new Path(path))))
line = br.readLine()
while (line != null) {
result += line
line = br.readLine()
}
} finally {
if (br != null) br.close()
}
result.mkString("n")
}
}
这一点在生产环境中还是有一定应用场景的,比如,通过解析mysql binlog日志,将数据同步到kafka,然后再通过消费者程序消费kafka中的数据保存到存储系统中,如delta,通过offset信息对比来校验,binlog到kafka的延迟(如,通过获取binlog中的offset信息与流程序同步到kafka时进行checkpoint的offset做对比)、kafka到存储系统中的延迟。
此外,要注意commits目录下记录的是已完成的批次信息。在实际进行offset比对时,要以此为基准再去获取offsets目录下的offsets信息。
- 联袂腾讯 “互联网+税务”创新模式在蒙启动
- 学医11年,终将被机器取代
- 大数据揭示:女性比男性更关注医改 建档等是热门话题
- 腾讯人工智能实验室AI Lab主任张潼博士前沿对话:AI如何助力营销?
- 当前所有源码链接
- Python入门基础连载(4)控制流
- 济西站构建大数据运营网络,打造智能化列车加工厂
- 五位数终端收购的域名dongxiao.cn已启用
- 全球互联网发展进入“拐点”——展望下一代互联网
- 2 分钟论文:语音生成表情包背后的技术原理
- 享学课堂谈-Python程序员的常见错误
- 区块链技术,如何提升网络安全?
- 趣店推“大白汽车”业务 启用域名dabaiqiche.com
- 糖果吃了那么多,你真的知道比特币分叉是咋回事吗?
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- Android 中ListView和GridView赋值错位
- 谈谈类加载器
- Android中AlarmManager+Notification实现定时通知提醒功能
- Java垃圾回收相关面试题
- Android中backgroundDimEnabled的作用
- 甲基化相关的习题背景补充
- Android 完全退出的实例详解
- Android 双击Back键退出应用的实现方法
- 2020最全Java面试题--基础篇
- Android开发之背景动画简单实现方法
- 15.深入k8s:Event事件处理及其源码分析
- 说一说 HashMap 底层数据结构
- 详解Android Service 使用时的注意事项
- Android自定义View实现QQ运动积分转盘抽奖功能
- 请简述Spring JDBC是如何进行配置的