本文共 631 字,大约阅读时间需要 2 分钟。
在spark streaming中更新数据到zookeeper,第一次更新的时候可能会报如上错(from kafka_0.9),
解决方案就是在测试过程中临时加入如下代码: 目的是创建相对应的文件夹
ZkUtils(zkClient, false).updatePersistentPath(path="/consumers/groupId", data="groupId")
ZkUtils(zkClient, false).updatePersistentPath(path="/consumers/groupId/offsets", data="offset") ZkUtils(zkClient, false).updatePersistentPath("/consumers/groupIdoffsets/topic1", "partiiton=12") ZkUtils(zkClient, false).updatePersistentPath("/consumers/groupId/offsets/topic2", "partition=3")// 提交部分
val zkPath = s"${topicDir.consumerOffsetDir}/${o.partition}"
ZkUtils(zkClient, isZkSecurityEnabled=false).updatePersistentPath(zkPath, offset)转载地址:http://zfzqj.baihongyu.com/