sparkstreaming+kafka+hbase实时写入数据，执行6天后，出现异常，数据无法写入

没找到怎么贴图片-------

tf.filter(!_.contains(str1)).filter(!_.contains(str2)).map(_.replaceAll(""", "")).map(_.split(",")).filter(_.length == 7)

.map(p => (p(0).substring(3), p(1), p(4), p(5), p(6))).map(line => {

val put = new Put(Bytes.toBytes(line._1.reverse + MD5Util.stringMD5(UUID.randomUUID().toString)))

put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("DeviceID"), Bytes.toBytes(line._1))

put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("DeviceName"), Bytes.toBytes(line._2))

put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("CollectionTime"), Bytes.toBytes(line._3))

put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("GranularityPeriod"), Bytes.toBytes(line._4))

put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("cpuUsage"), Bytes.toBytes(line._5))

}).foreachPartition(line => { val conn = ConnectionFactory.createConnection(HBaseConfiguration.create)

val table = conn.getTable(TableName.valueOf("eSight_cpu")) import scala.collection.JavaConversions._

table.put(seqAsJavaList(line.toSeq))

})

以上是代码。
以下是错误信息：

java.io.IOException: Bad response ERROR for BP-983455033-190.15.119.7-1525777063721:blk_1077688664_3948006
from datanode DatanodeInfoWithStorage[190.15.119.10:25009,DS-cc5cf8be-f909-4651-9afa-113776e46d16,DISK]
at org.apache.hadoop.hdfs.DataStreamer$ResponseProcessor.run(DataStreamer.java:1148)
然后就是一直：

[Stage 1998:============================> (1 + 1) / 2]

[Stage 1998:============================> (1 + 1) / 2]

[Stage 1998:============================> (1 + 1) / 2]
卡在这里。