图文精华

About云-梭伦科技»专题 › 交流区 › 技术交流 › Spark › 【求助】Spark如果在写入多个HBase表中
分享

【求助】Spark如果在写入多个HBase表中

唐运 发表于 2015-3-22 23:08:04 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 11 85191
【求助】Spark如果在写入多个HBase表中，最近项目需要用到Spark Streaming，在研究写入HBase时，单表OK。
但是写入多表没有成功，请有做过的朋友帮忙指点下。万分感谢

回复

使用道具 举报

自定义广告语

已有(11)人评论

跳转到指定楼层
正序浏览
langke93 发表于 2015-3-22 23:30:27
贴出代码和错误看看
回复

使用道具 举报

desehawk 发表于 2015-3-22 23:36:17
本帖最后由 desehawk 于 2015-3-22 23:45 编辑

单表能成功，多表应该也没有问题，是不是因为使用的参数或则变量等程序内部造成的。
回复

使用道具 举报

唐运 发表于 2015-3-23 00:41:25
本帖最后由 唐运 于 2015-3-23 00:43 编辑 
  1. package com.stream

  2. 

  3. import org.apache.spark.streaming.StreamingContext

  4. import org.apache.spark.streaming.Seconds

  5. import org.apache.spark.streaming.kafka.KafkaUtils

  6. import kafka.serializer.StringDecoder

  7. import com.utils.HBaseUtils

  8. import org.apache.hadoop.hbase.HBaseConfiguration

  9. import org.apache.hadoop.mapred.JobConf

  10. import org.apache.hadoop.hbase.mapred.TableOutputFormat

  11. import org.apache.spark.rdd.RDD

  12. import org.apache.spark.rdd.PairRDDFunctions

  13. import com.bean.LogFormatBean

  14. import scala.collection.mutable.Map

  15. import org.apache.hadoop.hbase.client.Put

  16. import org.apache.hadoop.hbase.io.ImmutableBytesWritable

  17. import org.apache.hadoop.hbase.util.Bytes

  18. import org.apache.hadoop.hbase.mapreduce.MultiTableOutputFormat

  19. 

  20. //import kafka.serializer.StringDecoder

  21. //import org.apache.spark.streaming._

  22. //import org.apache.spark.streaming.kafka._

  23. //import org.apache.hadoop.mapred.JobConf

  24. //import org.apache.spark.rdd.PairRDDFunctions

  25. //import org.apache.hadoop.hbase.util.Bytes

  26. 

  27. 

  28. 

  29. object KafkaToHBase {

  30.   def main(args: Array[String]): Unit = {

  31.     if (args.length < 6) {

  32.       System.err.println("Usage: KafkaToHBase <master> <zkQuorum> <group> <topics> <HBaseTableName> <brokers>")

  33.       System.exit(1)

  34.     }

  35. 

  36.     //提取参数

  37.     val Array(master, zkQuorum, group, topics, tableName, brokers) = args

  38.     

  39.      //设置批处理间隔2秒

  40.     val ssc = new StreamingContext(master, "SparkImportHBase", Seconds(1))

  41.     //提取topic、brokers参数

  42.     val topicsSet = topics.split(",").toSet//多topic同时接收,如果直写一个topic就只接收一个topic的数据

  43.     val kafkaParams = scala.collection.immutable.Map[String, String]("metadata.broker.list" -> brokers)

  44.     //接收Kafka传递的数据

  45.     val kafkaDstreams = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

  46.     //遍历rdd

  47.     val lines = kafkaDstreams.foreachRDD{rdd=>

  48.       //提rdd中的取值

  49.       val dataSet = rdd.map(_._2)

  50.       saveGameLogToHBase(dataSet, zkQuorum, tableName)

  51.     }

  52.     kafkaDstreams.map(_._2).count().print()

  53.     

  54.     ssc.start()

  55.     ssc.awaitTermination()

  56.   }

  57.   

  58.   

  59.   /**

  60.    * 游戏原始日志保存至HBase

  61.    */

  62. 

  63.    def saveGameLogToHBase(rdd: RDD[String], zkQuorum: String, tableName: String): Unit = {

  64.                   val conf = HBaseConfiguration.create()

  65.                  //设置zookeeper地址

  66.      conf.set("hbase.zookeeper.quorum", zkQuorum)

  67.       //设置表名,如果不是原始日志

  68.     val columnAndvalueMap = Map[String, String]()

  69.     new PairRDDFunctions(rdd.map { line =>

  70.       //生成rowkey

  71.         //val rowkey = HBaseUtils.createRowKey(line, tableName)

  72.       val rowkey = HBaseUtils.buildUUID

  73.         //val tablename = "n2_outer_"+line.split("#")(0)+"_log"

  74.         //conf.set(TableOutputFormat.OUTPUT_TABLE,tableName)

  75.         gameLogToMap("#", line, columnAndvalueMap)

  76.         val tablebname = line.split("#")(0)

  77. 

  78.       createHBaseRecord(LogFormatBean.CF, rowkey, columnAndvalueMap,tablebname)

  79. 

  80. //    }).saveAsHadoopDataset(jobConfig)

  81.   }).saveAsNewAPIHadoopFile("",classOf[String] , classOf[String], classOf[MultiTableOutputFormat], conf)

  82.   }

  83.    

  84.      def gameLogToMap(regex: String, line: String, mp: Map[String, String]) = {

  85.     val s_lines = line.split(regex)

  86.     val length: Int = s_lines.length

  87.     val mp = Map[String, String]()

  88.     //循环遍历参数

  89.     for (index <- 1 to length) {

  90.       mp.put("c" + index, s_lines(index - 1))

  91.     }

  92.     mp

  93.   }

  94.      

  95.      

  96.      def createHBaseRecord(colfimaly: String, rowKey: String, values: Map[String, String],tablebname:String): (ImmutableBytesWritable, Put) = {

  97.     //创建一个put 

  98.     val record = new Put(Bytes.toBytes(rowKey))

  99.     //遍历添加参数

  100.     val it = values.iterator

  101.     while (it.hasNext) {

  102.       val vls = it.next()

  103.       val colname = vls._1.toString

  104.       println("colname============>"+colname)

  105.       val colvalue = vls._2.toString

  106.       println("colvalue============>"+colvalue)

  107.       if(!colname.eq("") && !colvalue.eq(""))

  108.         record.add(Bytes.toBytes(colfimaly), Bytes.toBytes(colname), Bytes.toBytes(colvalue))

  109.       else

  110.          record.add(Bytes.toBytes(colfimaly), Bytes.toBytes("c1"), Bytes.toBytes("valueIsNull"))

  111.     }

  112.     //返回put

  113.     (new ImmutableBytesWritable(Bytes.toBytes(tablebname)), record)

  114.   }

  115.      

  116. }
复制代码
回复

使用道具 举报

唐运 发表于 2015-3-23 00:42:30
单表是按照官网实例来的，这个是根据api和源码之后写的，现在报错
  1. org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 290.0 failed 4 times, most recent failure: Lost task 3.3 in stage 290.0 (TID 1115, dn14.9961.tj): java.lang.IllegalArgumentException: No columns to insert

  2.         at org.apache.hadoop.hbase.client.HTable.validatePut(HTable.java:1007)

  3.         at org.apache.hadoop.hbase.client.HTable.doPut(HTable.java:800)

  4.         at org.apache.hadoop.hbase.client.HTable.put(HTable.java:786)

  5.         at org.apache.hadoop.hbase.mapreduce.MultiTableOutputFormat$MultiTableRecordWriter.write(MultiTableOutputFormat.java:132)

  6.         at org.apache.hadoop.hbase.mapreduce.MultiTableOutputFormat$MultiTableRecordWriter.write(MultiTableOutputFormat.java:68)

  7.         at org.apache.spark.rdd.PairRDDFunctions$anonfun$12.apply(PairRDDFunctions.scala:1000)

  8.         at org.apache.spark.rdd.PairRDDFunctions$anonfun$12.apply(PairRDDFunctions.scala:979)

  9.         at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)

  10.         at org.apache.spark.scheduler.Task.run(Task.scala:64)

  11.         at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)

  12.         at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)

  13.         at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)

  14.         at java.lang.Thread.run(Thread.java:662)

  15. 

  16. Driver stacktrace:

  17.         at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$failJobAndIndependentStages(DAGScheduler.scala:1203)

  18.         at org.apache.spark.scheduler.DAGScheduler$anonfun$abortStage$1.apply(DAGScheduler.scala:1192)

  19.         at org.apache.spark.scheduler.DAGScheduler$anonfun$abortStage$1.apply(DAGScheduler.scala:1191)

  20.         at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)

  21.         at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)

  22.         at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1191)

  23.         at org.apache.spark.scheduler.DAGScheduler$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:693)

  24.         at org.apache.spark.scheduler.DAGScheduler$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:693)

  25.         at scala.Option.foreach(Option.scala:236)

  26.         at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:693)

  27.         at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1393)

  28.         at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1354)
复制代码
回复

使用道具 举报

arsenduan 发表于 2015-3-23 01:01:05

问题可能在这里了，看看是不是循环或则边界造成的。先一一打印，colname很有可能因为边界造成找不到。
while (it.hasNext) {
      val vls = it.next()
      val colname = vls._1.toString
      println("colname============>"+colname)
      val colvalue = vls._2.toString
      println("colvalue============>"+colvalue)
      if(!colname.eq("") && !colvalue.eq(""))
       // record.add(Bytes.toBytes(colfimaly), Bytes.toBytes(colname), Bytes.toBytes(colvalue))
      else
         //record.add(Bytes.toBytes(colfimaly), Bytes.toBytes("c1"), Bytes.toBytes("valueIsNull"))
    }

回复

使用道具 举报

唐运 发表于 2015-3-23 01:01:27
今天真是晕倒家了，烦了超低级错误
正确代码如下：：
  1. package com.stream

  2. 

  3. import org.apache.spark.streaming.StreamingContext

  4. import org.apache.spark.streaming.Seconds

  5. import org.apache.spark.streaming.kafka.KafkaUtils

  6. import kafka.serializer.StringDecoder

  7. import com.utils.HBaseUtils

  8. import org.apache.hadoop.hbase.HBaseConfiguration

  9. import org.apache.hadoop.mapred.JobConf

  10. import org.apache.hadoop.hbase.mapred.TableOutputFormat

  11. import org.apache.spark.rdd.RDD

  12. import org.apache.spark.rdd.PairRDDFunctions

  13. import com.bean.LogFormatBean

  14. import scala.collection.mutable.Map

  15. import org.apache.hadoop.hbase.client.Put

  16. import org.apache.hadoop.hbase.io.ImmutableBytesWritable

  17. import org.apache.hadoop.hbase.util.Bytes

  18. import org.apache.hadoop.hbase.mapreduce.MultiTableOutputFormat

  19. 

  20. //import kafka.serializer.StringDecoder

  21. //import org.apache.spark.streaming._

  22. //import org.apache.spark.streaming.kafka._

  23. //import org.apache.hadoop.mapred.JobConf

  24. //import org.apache.spark.rdd.PairRDDFunctions

  25. //import org.apache.hadoop.hbase.util.Bytes

  26. 

  27. 

  28. 

  29. object KafkaToHBase {

  30.   def main(args: Array[String]): Unit = {

  31.     if (args.length < 6) {

  32.       System.err.println("Usage: KafkaToHBase <master> <zkQuorum> <group> <topics> <HBaseTableName> <brokers>")

  33.       System.exit(1)

  34.     }

  35. 

  36.     //提取参数

  37.     val Array(master, zkQuorum, group, topics, tableName, brokers) = args

  38.     

  39.      //设置批处理间隔2秒

  40.     val ssc = new StreamingContext(master, "SparkImportHBase", Seconds(1))

  41.     //提取topic、brokers参数

  42.     val topicsSet = topics.split(",").toSet//多topic同时接收,如果直写一个topic就只接收一个topic的数据

  43.     val kafkaParams = scala.collection.immutable.Map[String, String]("metadata.broker.list" -> brokers)

  44.     //接收Kafka传递的数据

  45.     val kafkaDstreams = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

  46.     //遍历rdd

  47.     val lines = kafkaDstreams.foreachRDD{rdd=>

  48.       //提rdd中的取值

  49.       val dataSet = rdd.map(_._2)

  50.       saveGameLogToHBase(dataSet, zkQuorum, tableName)

  51.     }

  52.     kafkaDstreams.map(_._2).count().print()

  53.     

  54.     ssc.start()

  55.     ssc.awaitTermination()

  56.   }

  57.   

  58.   

  59.   /**

  60.    * 游戏原始日志保存至HBase

  61.    */

  62. 

  63.    def saveGameLogToHBase(rdd: RDD[String], zkQuorum: String, tableName: String): Unit = {

  64.       val conf = HBaseConfiguration.create()

  65.      //设置zookeeper地址

  66.      conf.set("hbase.zookeeper.quorum", zkQuorum)

  67.       //设置表名,如果不是原始日志

  68.     new PairRDDFunctions(rdd.map { line =>

  69.       //生成rowkey

  70.         //val rowkey = HBaseUtils.createRowKey(line, tableName)

  71.       val rowkey = HBaseUtils.buildUUID

  72.         //val tablename = "n2_outer_"+line.split("#")(0)+"_log"

  73.         //conf.set(TableOutputFormat.OUTPUT_TABLE,tableName)

  74.         val columnAndvalueMap = gameLogToMap("#", line)

  75.         val tablebname = line.split("#")(0)

  76. 

  77.       createHBaseRecord(LogFormatBean.CF, rowkey, columnAndvalueMap,tablebname)

  78. 

  79. //    }).saveAsHadoopDataset(jobConfig)

  80.   }).saveAsNewAPIHadoopFile("",classOf[String] , classOf[String], classOf[MultiTableOutputFormat], conf)

  81.   }

  82.    

  83.      def gameLogToMap(regex: String, line: String) = {

  84.     val s_lines = line.split(regex)

  85.     val length: Int = s_lines.length

  86.     val mp = Map[String, String]()

  87.     //循环遍历参数

  88.     for (index <- 1 to length) {

  89.       mp.put("c" + index, s_lines(index - 1))

  90.     }

  91.     mp

  92.   }

  93.      

  94.      

  95.      def createHBaseRecord(colfimaly: String, rowKey: String, values: Map[String, String],tablebname:String): (ImmutableBytesWritable, Put) = {

  96.     //创建一个put 

  97.     val record = new Put(Bytes.toBytes(rowKey))

  98.     //遍历添加参数

  99.     val it = values.iterator

  100.     while (it.hasNext) {

  101.       val vls = it.next()

  102.       val colname = vls._1.toString

  103.       println("colname============>"+colname)

  104.       val colvalue = vls._2.toString

  105.       println("colvalue============>"+colvalue)

  106.       record.add(Bytes.toBytes(colfimaly), Bytes.toBytes("c1"), Bytes.toBytes("valueIsNull"))

  107.     }

  108.     //返回put

  109.     (new ImmutableBytesWritable(Bytes.toBytes(tablebname)), record)

  110.   }

  111.      

  112. }
复制代码
回复

使用道具 举报

arsenduan 发表于 2015-3-23 01:03:01
本帖最后由 arsenduan 于 2015-3-23 01:04 编辑
唐运 发表于 2015-3-23 00:42
单表是按照官网实例来的，这个是根据api和源码之后写的，现在报错

也不能只看columns ，还需要看看是否有空值，可能put里面 有空值了
查看
HBase-No columns to insert
回复

使用道具 举报

唐运 发表于 2015-3-23 08:12:50
arsenduan 发表于 2015-3-23 01:03
也不能只看columns ，还需要看看是否有空值，可能put里面 有空值了
查看
HBase-No columns to insert

就是因为空值，一个方法有返回值，我没有接收，写hbase时用的Map是个空的。

回复

使用道具 举报

落魂草 发表于 2015-3-23 19:43:54
回复

使用道具 举报

12下一页
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
唐运

中级会员

关注

  • 17

    主题

  • 41

    帖子

  • 4

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2025 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000