通过 Spark R 操作 Hive

作为数据工程师,我日常用的主力语言是R,HiveQL,Java与Scala。R是非常适合做数据清洗的脚本语言,并且有非常好用的服务端IDE——RStudio Server;而用户日志主要储存在hive中,因此用HiveQL写job也是日常工作之一;当然R的执行效率确实不靠谱,因此还需要Java(Elasticsearch,Hadoop的原生语言)与Scala(Spark的原生语言)的帮助。

R和python一样也是一个很好的胶水语言,在搜索引擎的实战中,我就是用R来完成了ES集群索引的全量/增量更新操作。既然是一瓶胶水,你很难不希望它粘合上越来越多的东西。对于一个有处女座倾向的双鱼座,因为RHive的安装和配置太复杂而一直没法把R和Hive粘合起来是我长期以来的苦恼。最近要用Spark的Mlib做文本分类工作于是开始重新鼓捣Spark,没想到歪打正着地发现用Spark的R语言api可以操作Hive,搞通之后真是感到Awesome as fuck!

一、编译安装带有hive与R的Spark

官网上预编译好的spark都是不带hive支持的,因此只能自己从源码重新编译(在这里要感谢spark老司机同事的指点)。

# 从官网下载源码包
wget http://mirrors.hust.edu.cn/apache/spark/spark-1.6.1/spark-1.6.1.tgz

# 解压
tar zxvf spark-1.6.1.tgz

# 为编译R的支持做准备
spark-1.6.1/R/install-dev.sh

# 编译安装带有hive与R的spark
spark-1.6.1/make-distribution.sh --name hadoop2.5 --skip-java-test --tgz -Psparkr -Pyarn -Dhadoop.version=2.5.0 -Dscala-2.10.4 -Phive -Phive-thriftserver

# 把编译好的包解压到自己习惯的目录(参考elasticsearch放在/usr/share/目录下)
tar zxvf spark-1.6.1/spark-1.6.1-bin-hadoop2.5.tgz
mv spark-1.6.1/spark-1.6.1-bin-hadoop2.5 /usr/share/

二、环境配置

环境配置可以直接参考之前的文章CentOS 6.7 hadoop free版本Spark 1.6安装与使用,此外要添加一下spark-defaults.conf这个配置文件,以使得hive能够被顺利连接上:

# vim $SPARK_HOME/conf/spark-defaults.conf

spark.yarn.jar=hdfs://n1:8020/user/spark/share/lib/spark-assembly-1.6.1-hadoop2.5.0.jar
spark.driver.extraLibraryPath=/opt/cloudera/parcels/CDH-5.3.0-1.cdh5.3.0.p0.30/lib/hadoop/lib/native
spark.executor.extraLibraryPath=/opt/cloudera/parcels/CDH-5.3.0-1.cdh5.3.0.p0.30/lib/hadoop/lib/native

# 注:需要把$SPARK_HOME/lib/spark-assembly-1.6.1-hadoop2.5.0.jar这个文件放到hdfs相应的目录下

好吧,这一步也是完全按照同事之前的经验整理的,还没有深入研究到底是如何生效的,但重要的是先用起来……

三、读写hive实例

直接上代码:

Sys.setenv(SPARK_HOME = "/usr/share/spark-1.6.1-bin-hadoop2.5", HADOOP_CONF_DIR="/opt/cloudera/parcels/CDH/lib/hadoop/etc/hadoop")
library(SparkR, lib.loc = c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib")))

sc <- sparkR.init(master = "yarn-client", appName = "test")
hiveContext <- sparkRHive.init(sc)

# 读取数据
results <- sql(hiveContext, "select * from dd_b_ec_e46_clickmodel where dt=20150601")
head(results)

# 写入数据
saveAsTable(results, "tmp_test")

这里需要注意的是:

  • 如果运行R的用户不是hdfs管理员用户的话,可以把运行R的用户添加到可访问hdfs的用户组里
  • 另外加上HADOOP_USER_NAME=”hdfs”这个环境变量(假设hive的所有者是hdfs)就可以进行写表操作了

总结

大致看了一遍Spark R api后,还是略感失望的,因为发现Spark当前对R的支持主要就是集中在dataframe部分,连最基本的map和reduce操作都没有,看来也只是把R当成数据清洗的工具了。不知不觉中砸了两周的时间在Spark上,接下俩得抖擞精神做一下业务需求了,然后就是硬刚Scala和Mlib!

发表评论

电子邮件地址不会被公开。 必填项已用*标注