This seems a bug of spark, see https://github.com/apache/spark/pull/16325 my code is here,it's work.But after a while,there are a lot of .hive-stagingXXXX directory. a photo of my warehouse as attachment. val sparkConf = new SparkConf() .setAppName("SendHistoryConsumer") // .setMaster("local[2]") // for debug val ssc = new StreamingContext(sparkConf, Seconds(batchInterval.toInt)) ssc.checkpoint("/opt/cloudera/checkPoint") // kafka configuration val topicsSet = Set(KafkaConfigConstant.PRIORITY_10) val kafkaParams = Map[String, String]( "metadata.broker.list" -> brokers, "serializer.class"->"kafka.serializer.StringEncoder", "group.id" -> groupId, "auto.offset.reset" -> "smallest" val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet) //messages.print() val dStream = messages .map(_._2) .filter(message => !message.isEmpty ) .map(parseTopic10) dStream.foreachRDD(rdd => if (!rdd.isEmpty && !rdd.partitions.isEmpty) { val hiveCtx = new HiveContext(rdd.sparkContext) val sendCallBackPoolDF = hiveCtx.createDataFrame(rdd, classOf[SmsSendHistDataSede]) hiveCtx.sql("use db") sendCallBackPoolDF.registerTempTable("sendCallBack") hiveCtx.sql("set hive.exec.dynamic.partition=true") hiveCtx.sql("set hive.exec.dynamic.partition.mode=nonstrict") val smTrgPart = hiveCtx.sql("insert into table SmsSendHistData_orc partition" + "(partittime) select userid,city,mobile,msgcount,msgbody,sender,accepttype,sendcode," + "msgid,bmsgid,usermsgid,userbulkmsgid,channelmsgid,channelid,fileid,notifyurl,smscode ,smsstat," + "notifycallnum,sendsmsbody,parentmsgid,errormessage,queuename,startdeliveryid,acctime,sendtime," + "notifytime,smstype,from_unixtime(floor(acctime/1000),'yyyyMMdd') as partittime from sendCallBack") smTrgPart.saveAsTable("SmsSendHistData_orc", SaveMode.Append) }
... View more
