关于spark-hbase在pyspark上的那些破事

2023年7月7日 298次阅读 来源: 咩咩红莉栖

背景:

项目需要读取Hbase并把计算结果保存在Hbase里供其他接口获取。

算法由pyspark实现。

原先Hbase的Thrift接口三天两头宕，而且性能低下。

充满糟点的background结束

结论：

先说结论，想节约时间的可以跳过后面的”充满糟点过程”部分。

前提：

1. 你要有个可以通repo的spark集群，或通过某种手段可以通repo（自建内网 http://repo.hortonworks.com/content/groups/public/ 的clone，或proxy）

2. spark-shell/spark-submit部分：

spark-submit \
--conf "spark.driver.extraJavaOptions=-Dhttp.proxyHost=<代理服务器IP> \
-Dhttp.proxyPort=<代理服务器端口> -Dhttps.proxyHost=<代理服务器IP> \
-Dhttps.proxyPort=<代理服务器端口>"\
--packages com.hortonworks:shc-core:1.1.1-2.1-s_2.11 \
--repositories http://repo.hortonworks.com/content/groups/public/ \
--files files:////hbase-site.xml script.py arg1, arg2

相关的lib依赖会自动下载, 如果实在没有手段通外网, 可以考虑先用此命令在通外网的机器上产生ivy2的cache,然后复制到master服务器上(client只需要master有相关lib即可)。

配置及注意点:

此方法未在Spark Yarn/Yarn Client 模式下测试，有童鞋搞定了且愿意分享的话可以补充。

代理只能HTTP或HTTPS协议(自己用polipo转一个就可以,不赘述)。

Local及Standalone模式经测试无问题。

由于用的是shc-core，因此保险起见推荐将hbase-site.xml复制进$SPARK_HOME/conf, 之前直接-files里提交文件，结果查库的时候连接失败了（emmmm）

用pyspark DataFrame操作Hbase:

和JAVA/Scala版的shc一样，先要定义catalog

catalog = ''.join("""{
  "table":{"namespace":"test", "name":"test_table"},
  "rowkey":"key",
  "columns":{
  "col0":{"cf":"rowkey", "col":"key", "type":"string"},
  "col1":{"cf":"result", "col":"class", "type":"string"}
  }
  }""".split())

造一个dataframe来测试写入：

  data_source_format = 'org.apache.spark.sql.execution.datasources.hbase'
  df = sc.parallelize([('a', '1.0'), ('b', '2.0')]).toDF(schema=['col0', 'col1'])
  df.show()
  df.write.options(catalog=catalog,newTable="5").format(data_source_format).save()

注: 推荐写入时加上”newTable”选项，否则当表不存在时会报如下错误，也就是提醒你新表至少要有3个regions：

Py4JJavaError: An error occurred while calling o510.save.: 
org.apache.spark.sql.execution.datasources.hbase.InvalidRegionNumberException:
Number of regions specified for new table must be greater than 3.

再读取数据:

df_read = spark.read.options(catalog=catalog).format(data_source_format).load()
df_read.show()

注: 读取的时候可以在load()之后加上各种select where 语句, 会自动转换为各种不人性的scan filters，并延迟加载到读取时执行。注意，因为spark是lazy执行的，如果where特别复杂的话推荐先load().cache()完之后接一句dataframe.count()再进行where处理，否则语句会被转换为各种filter，并在Hbase中处理，value filter的速度有目共睹…当然Hbase集群够强也可以无视。

充满糟点的过程：

在Thrift接口N**N次 OOM 之后，终于忍无可忍的想直接用原生spark-hbase读写数据。

OK，先去Hbase官网上找了reference，关于spark的部分只有Java和Scala有木有？spark-hbase项目不知死活有木有？最可气的是reference上那个版本号（带beta）全宇宙的repo都找不到有木有？github只有个空页面有木有？
虽然是开源项目，好歹reference也稍微上点心好不好-_-||。
（未完待续）

    原文作者：咩咩红莉栖
    原文地址: https://www.jianshu.com/p/0e25f4e10cdf
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞