RDD 只有第一列值：Hbase，PySpark

Posted 2023-04-15

【中文标题】RDD 只有第一列值：Hbase，PySpark

【英文标题】：RDD is having only first column value : Hbase, PySpark

【发布时间】：2015-06-23 17:52:12

【问题描述】：

我们正在使用以下命令使用 Pyspark 读取 Hbase 表。

from pyspark.sql.types import *
host=<Host Name>
port=<Port Number>

keyConv = "org.apache.spark.examples.pythonconverters.ImmutableBytesWritableToStringConverter"
valueConv = "org.apache.spark.examples.pythonconverters.HBaseResultToStringConverter"

cmdata_conf = "hbase.zookeeper.property.clientPort":port, "hbase.zookeeper.quorum": host, "hbase.mapreduce.inputtable": "CMData", "hbase.mapreduce.scan.columns": "info:Tenure info:Age"

cmdata_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=cmdata_conf)

output = cmdata_rdd.collect()

output

我得到的结果如下。 （钥匙和年龄）

[(u'123', u'5'), (u'234', u'4'), (u'345', u'3'), (u'456', u'4'), (u'567', u'7'), (u'678', u'7'), (u'789', u'8')]

相反，我期待 Key、Tenure 和 Age。如果我只有 Tenure 列，那么它会返回 Key 和 Tenure。但如果添加更多列，结果总是有 Key 和 Age 列。

谁能帮我们解决这个问题？

注意：我们是这个工具的新手

提前谢谢你。

【问题讨论】：

这个例子可能会有所帮助gist.github.com/2353b5da5099e24da111.git如果您还有任何问题请联系我

嗨，Anant，感谢您的回复，但是当我点击该链接时，我只能看到一个空白页:(

抱歉你需要 git clone 或者你也可以使用gist.github.com/anantasty/2353b5da5099e24da111

当我尝试从 HBase 查询多个列时遇到了同样的问题。我相信使用valueConv = "org.apache.spark.examples.pythonconverters.HBaseResultToStringConverter" 是它出错的地方。我们应该从 HBase 中获取一个列表 / 可迭代对象作为值，但不知何故，只有这些 Java 可迭代对象的最后一个条目被转换为 python 字符串。这应该是一张错误票。

【参考方案1】：

如果您正在制作原型并且不想更新您的集群，看看 happybase (https://happybase.readthedocs.org/en/latest/) 会很有用。

以下代码可以在一秒钟内从我的集群中获取我的小型 (9Gig) Hbase 表“name_Hbase_Table”。

import happybase
connection = happybase.Connection(host ='your.ip.cluster') #don't specify :port
table = connection.table('name_Hbase_Table')
def hbaseAccelerationParser(table): #create UDF to format data
    finalTable=[]
    for key, data in table.scan(): #don't need the key in my case
        line=[]
        for values in data.itervalues():
            line.append(values)
        finalTable.append(line)
    return finalTable
table =table.map(hbaseAccelerationParser) #capture data in desired format
table = sc.parallelize(table ,4) #put in RDD

【讨论】：

这真的有效吗？我在 table =table.map(hbaseAccelerationParser) #capture data in desired format Traceback（最近一次调用最后一次）上出现错误：文件“”，第 1 行，在 AttributeError：'Table' 对象没有属性“地图”