1 問(wèn)題描述

在使用Spark BulkLoad數(shù)據(jù)到HBase時(shí)遇到以下問(wèn)題：

17/05/19 14:47:26 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 12.0 (TID 79, bydslave5, executor 3): java.io.IOException: Non-increasing Bloom keys: 80a01055HAXMTXG10100001KEY_VOLTAGE_T_C_PWR after af401055HAXMTXG10100001KEY_VOLTAGE_TEC_PWR
    at org.apache.hadoop.hbase.regionserver.StoreFile$Writer.appendGeneralBloomfilter(StoreFile.java:911)
    at org.apache.hadoop.hbase.regionserver.StoreFile$Writer.append(StoreFile.java:947)
    at org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2$1.write(HFileOutputFormat2.java:199)
    at org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2$1.write(HFileOutputFormat2.java:152)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply$mcV$sp(PairRDDFunctions.scala:1125)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply(PairRDDFunctions.scala:1123)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply(PairRDDFunctions.scala:1123)
    at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1341)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1131)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1102)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:99)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

那么是在什么時(shí)候出現(xiàn)的呢？在運(yùn)行完下面語(yǔ)句

val rdd = sc.textFile("/data/produce/2015/service.log.2017-04-24-08").map(_.split("@")).map{x => (DigestUtils.md5Hex(x(0)+x(1)).substring(0,3)+x(0)+x(1),x(2))}.map{x=>{val kv:KeyValue = new KeyValue(Bytes.toBytes(x._1),Bytes.toBytes("v"),Bytes.toBytes("value"),Bytes.toBytes(x._2+""));(new ImmutableBytesWritable(kv.getKey),kv)}}

rdd.saveAsNewAPIHadoopFile("/tmp/data1",classOf[ImmutableBytesWritable],classOf[KeyValue],classOf[HFileOutputFormat],job.getConfiguration())

從報(bào)錯(cuò)信息來(lái)看是因?yàn)閗ey沒(méi)有按照遞增的順序進(jìn)行排列，可能是BloomFilter對(duì)key的排序有要求，但是我們知道key的無(wú)序是因?yàn)閟park在shuffle階段并沒(méi)有像MapReduce那樣強(qiáng)制排序，所以要解決這個(gè)問(wèn)題我們需要手動(dòng)地為數(shù)據(jù)進(jìn)行排序，只需要對(duì)rdd執(zhí)行sortBy即可。

2 問(wèn)題解決

下面語(yǔ)句是增加排序的語(yǔ)句，經(jīng)過(guò)測(cè)試運(yùn)行通過(guò)

val rdd = sc.textFile("/data/produce/2015/service.log.2017-04-24-08").map(_.split("@")).map{x => (DigestUtils.md5Hex(x(0)+x(1)).substring(0,3)+x(0)+x(1),x(2))}.sortBy(x =>x._1).map{x=>{val kv:KeyValue = new KeyValue(Bytes.toBytes(x._1),Bytes.toBytes("v"),Bytes.toBytes("value"),Bytes.toBytes(x._2+""));(new ImmutableBytesWritable(kv.getKey),kv)}}

rdd.saveAsNewAPIHadoopFile("/tmp/data1",classOf[ImmutableBytesWritable],classOf[KeyValue],classOf[HFileOutputFormat],job.getConfiguration())