Spark RDD按键查找

问题描述 投票:3回答:1

我有一个从HBase转换而来的RDD:

val hbaseRDD:RDD [(String,Array [String])]其中tuple._1是rowkey。并且数组是HBase中的值。

4929101-ACTIVE, ["4929101","2015-05-20 10:02:44","dummy1","dummy2"]
4929102-ACTIVE, ["4929102","2015-05-20 10:02:44","dummy1","dummy2"]
4929103-ACTIVE, ["4929103","2015-05-20 10:02:44","dummy1","dummy2"]

我还将SchemaRDD(id,date1,col1,col2,col3)转换为

val refDataRDD:RDD [(String,Array [String])]我将迭代并检查它是否存在于hbaseRDD中:

4929103, ["2015-05-21 10:03:44","EV01","col2","col3"]
4929104, ["2015-05-21 10:03:44","EV02","col2","col3"]

问题是,

  • 如何检查hbaseRDD中是否存在密钥(tuple._1)/(“4929103”)并获取相应的值(tuple._2)? - 我不能在rdd.filter中使用PairRDD的查找函数,它抛出“scala.MatchError:null”,但它在外面工作 val filteredRDD = rdd.filter(sqlRow => { val hbaseLookup = hbaseRDD.lookup(sqlRow(0).toString + "-ACTIVE") // if found, check if date1 of hbaseRDD < sqlRow(1) // else if not found, retain row true }) 我不确定这是不是问题,因为当我将查询行切换为:时,我也遇到了NPE: val sqlRowHbase = hbaseRDD.filter(row => { 注意:我在这些行之前做了hbaseRDD.count。和hbaseRDD.lookup在rdd.filter之外正常工作

所以基本上,我试图通过hbaseRDD中的键“找到”并获取行/值。加入它们有点复杂,因为两个RDD中的某些值可能为空。并且它取决于很多场景,哪些行将保留哪些数据。

scala apache-spark mapreduce hbase rdd
1个回答
0
投票

假设您需要查找的a_id集包含在RDD中,我认为您可以使用leftOuterJoin而不是迭代并查找每个值。

我在上面看到了关于date1潜在可变位置的评论。我不是在下面解决它,我认为这应该在查找本身之前通过每行的某种特定映射来处理。

如果我正确获得伪代码,你有一个(id, date)的RDD,并希望通过在hbase中查找数据来更新它,如果在hbase中找到该行的这个id并且它的日期早于refData中的日期,则更新日期。那是对的吗?

如果是这样,假设你有一些像这样的ref数据:

val refData = sc.parallelize(Array(
 ("4929103","2015-05-21 10:03:44"),
 ("4929104","2015-05-21 10:03:44")
))

还有来自Hbase的一些行数据:

val hbaseRDD = sc.parallelize(Array(
    ("4929101-ACTIVE", Array("4929101","2015-05-20 10:02:44")),
    ("4929102-ACTIVE", Array("4929102","2015-05-20 10:02:44")),
    ("4929103-ACTIVE", Array("4929103","2015-05-20 10:02:44"))
))

然后,您可以使用简单的leftOuterJoin从refData查找每个id到hbase,并且对于找到的每一行:如有必要,请更新日期:

refData
  // looks up in Hbase all rows whose date1 a_id value matches the id in searchedIds
  .leftOuterJoin(hbaseRDD.map{ case (rowkey, Array(a_id, date1)) => (a_id, date1)})

  // update the date in refData if date from hBase is earlier
  .map { case (rowKey, (refDate, maybeRowDate)) => ( rowKey, chooseDate (refDate, maybeRowDate)) }
  .collect


def chooseDate(refDate: String, rowDate: Option[String]) =  rowDate match {

  // if row not found in Hbase: keep ref date
  case None => refDate

  case Some(rDate) => 
    if (true) /* replace this by first parsing the date, then check if rowDate < refDate */ 
        rowDate
    else
        refDate
}
热门问题
推荐问题
最新问题