从 Spark Streaming 写入 HBase

Question

我在 Spark Streaming（使用 Scala）中有一个由（键，值）对组成的文件输入，如果键满足某个条件，我需要做的是将值存储在 HBase 中。 因为我有：

val pair: DStream[(String, String)]

我试图做的是地图中的一个条件，然后尝试在 HBase 中插入值：

pair.map(x => {
if (x._1 == "condition")
{ val hconf = HBaseConfiguration.create()
val hTable = new HTable(hconf, "mytab")
val thePut = new Put(Bytes.toBytes(1))
thePut.add(Bytes.toBytes("colfamily"), Bytes.toBytes("c1"), Bytes.toBytes(x._2)
hTable.put(thePut)
})
}

但是这不起作用，并且在使用 spark-submit 执行时出现错误提示：没有注册输出操作，因此没有执行任何操作

这是我能想到的将值插入 HBase 的唯一方法，我做错了什么吗？你能帮我修一下吗？

这是更新后的代码：

pair.foreachRDD(rdd => rdd.map( p =>
{val hconf = HBaseConfiguration.create()
 val hTable = new HTable(hconf,"mytab")
 val thePut = new Put(Bytes.toBytes(1))
 thePut.add(Bytes.toBytes("colfamily"), Bytes.toBytes(p._1), Bytes.toBytes(p._2)
 hTable.put(thePut)
})

当我使用 Spark-submit 运行它时，我收到一条错误消息“Task not serializable”，你知道这是什么意思吗？我该如何解决它？

提前谢谢你

Answer 1

让，

您收到错误消息的原因是您在代码中缺少对名为 pair 的 RDD 的操作。

请参阅下面更正的代码。

  pair.foreachRDD((rdd : [RDD(String,String)]) => {  
 val newRdd = rdd.map(p=> (p._1,p._2))
        if (mewRdd._1 == "condition")
        {
         /* Your code*/
        }})

要从 spark 流中插入数据到 hbase，请参阅这篇文章 https://www.mapr.com/blog/spark-streaming-hbase 我希望这会有所帮助。