在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)