上面一看代码就不少错误,不知道是从哪看到的,还是自己写的。 首先put.addColumn这个是没有 错误的,肯定是缺包或则前面的参数不对。建议参考标准的资料。 这里给你推荐下官网的资料 http://hbase.apache.org/book.html#hbase_apis 贴出官网跟你类似的例子 [mw_shl_code=scala,true]val sc = new SparkContext("local", "test") val config = new HBaseConfiguration() val hbaseContext = new HBaseContext(sc, config) val ssc = new StreamingContext(sc, Milliseconds(200)) val rdd1 = ... val rdd2 = ... val queue = mutable.Queue[RDD[(Array[Byte], Array[(Array[Byte], Array[Byte], Array[Byte])])]]() queue += rdd1 queue += rdd2 val dStream = ssc.queueStream(queue) dStream.hbaseBulkPut( hbaseContext, TableName.valueOf(tableName), (putRecord) => { val put = new Put(putRecord._1) putRecord._2.foreach((putValue) => put.addColumn(putValue._1, putValue._2, putValue._3)) put })[/mw_shl_code] 里面介绍的很详细 |
看下自己的参数是否正确,然后包是否引入 |