gpt4 book ai didi

scala - Spark RDD 按键查找

转载 作者:行者123 更新时间:2023-12-02 04:49:09 25 4
gpt4 key购买 nike

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

val hbaseRDD: RDD[(String, Array[String])] 其中 tuple._1 是行键。数组是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 中的某些值可能为空。这取决于很多场景,哪些行会保留哪些数据。

最佳答案

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

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

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

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

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
}

关于scala - Spark RDD 按键查找,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/30421484/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com