Spark:如何使用HBase过滤器，例如python-api的QualiferFilter

问题描述

我想通过使用python-api中的QualiferFilter之类的过滤器从HBase中获取行.
我知道从HBase获取行的方式，就像在代码下一样.

I want to get rows from HBase by using filter like QualiferFilter at python-api.
I know the way to get rows from HBase like under the code.

host = 'localhost'
keyConv = "org.apache.spark.examples.pythonconverters.ImmutableBytesWritableToStringConverter"
valueConv = "org.apache.spark.examples.pythonconverters.HBaseResultToStringConverter"
conf = {"hbase.zookeeper.quorum": host, "hbase.mapreduce.inputtable": "user",
                "hbase.mapreduce.scan.columns": "u:uid",
                "hbase.mapreduce.scan.row.start": "1", "hbase.mapreduce.scan.row.stop": "100"}
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)

但是，我也想通过使用过滤器来获取行.
我需要添加什么类型的代码?

But, I also want to get rows by using filters.
What type of codes do I need to add?

推荐答案

您可以检查此代码.......

Hello you can check this code................

def  doYourStuff(row):
     text = row.split("\n")
     data = {} 
     for row in text:
        if json.loads(row)["qualifier"] == "message":
              data["message"] = json.loads(row)["value"]
        if json.loads(row)["qualifier"] == "domain":
              data["domain"] = json.loads(row)["value"]
        data["rowKey"] = json.loads(row)["row"]
      return DoWhatYouWantToDo(data)

    def save_record(rdd):
        host = '172.31.@@.@@'
        table = 'TableName'
        keyConv1 = "org.apache.spark.examples.pythonconverters.StringToImmutableBytesWritableConverter"
        valueConv1 = "org.apache.spark.examples.pythonconverters.StringListToPutConverter"
        conf = {"hbase.zookeeper.quorum": host,
                "hbase.mapred.outputtable": table,
                "mapreduce.outputformat.class": "org.apache.hadoop.hbase.mapreduce.TableOutputFormat",
                "mapreduce.job.output.key.class": "org.apache.hadoop.hbase.io.ImmutableBytesWritable",
                "mapreduce.job.output.value.class": "org.apache.hadoop.io.Writable"}
         rdd.saveAsNewAPIHadoopDataset(
            keyConverter=keyConv1, valueConverter=valueConv1,conf=conf)


    hbaseRdd = hbaseRdd.map(lambda x: x[1])  # message_rdd = hbase_rdd.map(lambda x:x[0]) will give only row-key

    processedRdd = hbaseRdd.map(lambda x: doYourStuff(x))
    save_record(processedRdd)

这篇关于Spark:如何使用HBase过滤器，例如python-api的QualiferFilter的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！