python – 当使用hbase作为数据源时,spark是否使用hbase键的排序顺序

2023年5月23日 263次阅读

我将时间序列数据存储在HBase中. rowkey由user_id和timestamp组成,如下所示：

{
    "userid1-1428364800" : {
        "columnFamily1" : {
            "val" : "1"
            }
        }
    }
    "userid1-1428364803" : {
        "columnFamily1" : {
            "val" : "2"
            }
        }
    }

    "userid2-1428364812" : {
        "columnFamily1" : {
            "val" : "abc"
            }
        }
    }

}

现在我需要执行每用户分析.这是hbase_rdd的初始化(从here开始)

sc = SparkContext(appName="HBaseInputFormat")

conf = {"hbase.zookeeper.quorum": host, "hbase.mapreduce.inputtable": table}
keyConv = "org.apache.spark.examples.pythonconverters.ImmutableBytesWritableToStringConverter"
valueConv = "org.apache.spark.examples.pythonconverters.HBaseResultToStringConverter"

hbase_rdd = sc.newAPIHadoopRDD(
        "org.apache.hadoop.hbase.mapreduce.TableInputFormat",
        "org.apache.hadoop.hbase.io.ImmutableBytesWritable",
        "org.apache.hadoop.hbase.client.Result",
        keyConverter=keyConv,
        valueConverter=valueConv,
        conf=conf)

类似于自然mapreduce的处理方式是：

hbase_rdd
   .map(lambda row: (row[0].split('-')[0], (row[0].split('-')[1], row[1])))  # shift timestamp from key to value
   .groupByKey()
   .map(processUserData)  # process user's data

在执行第一个映射(从键到值的移位时间戳)时,知道当前用户的时间序列数据何时完成并因此可以启动groupByKey转换是至关重要的.因此,我们不需要映射所有表并存储所有临时数据.这是可能的,因为hbase按排序顺序存储行键.

使用hadoop流式传输可以通过以下方式完成：

import sys

current_user_data = []
last_userid = None
for line in sys.stdin:
    k, v = line.split('\t')
    userid, timestamp = k.split('-')
    if userid != last_userid and current_user_data:
        print processUserData(last_userid, current_user_data)
        last_userid = userid
        current_user_data = [(timestamp, v)]
    else:
        current_user_data.append((timestamp, v))

问题是：如何在Spark中使用hbase键的排序顺序？

最佳答案 我不太熟悉你从HBase中提取数据的方式所带来的保证,但如果我理解正确的话,我可以用普通的Spark来回答.

你有一些RDD [X].就Spark而言,RDD中的X是完全无序的.但是您有一些外部知识,并且您可以保证数据实际上是按X的某个字段分组(甚至可能按另一个字段排序).

在这种情况下,您可以使用mapPartitions执行与hadoop流媒体相同的操作.这使您可以迭代一个分区中的所有记录,因此您可以查找具有相同密钥的记录块.

val myRDD: RDD[X] = ...
val groupedData: RDD[Seq[X]] = myRdd.mapPartitions { itr =>
  var currentUserData = new scala.collection.mutable.ArrayBuffer[X]()
  var currentUser: X = null
  //itr is an iterator over *all* the records in one partition
  itr.flatMap { x => 
    if (currentUser != null && x.userId == currentUser.userId) {
      // same user as before -- add the data to our list
      currentUserData += x
      None
    } else {
      // its a new user -- return all the data for the old user, and make
      // another buffer for the new user
      val userDataGrouped = currentUserData
      currentUserData = new scala.collection.mutable.ArrayBuffer[X]()
      currentUserData += x
      currentUser = x
      Some(userDataGrouped)
    }
  }
}
// now groupedRDD has all the data for one user grouped together, and we didn't
// need to do an expensive shuffle.  Also, the above transformation is lazy, so
// we don't necessarily even store all that data in memory -- we could still
// do more filtering on the fly, eg:
val usersWithLotsOfData = groupedRDD.filter{ userData => userData.size > 10 }

我意识到你想使用python – 抱歉,我认为如果我用Scala编写,我更有可能得到正确的例子.我认为类型注释使意义更清晰,但它可能是斯卡拉偏见…… :).无论如何,希望你能理解发生的事情并将其翻译出来. (不要过分关注flatMap& Some& None,如果你理解这个想法可能并不重要……)

点赞