通过 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
mvspark-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.eventLog.dir=hdfs://n1:8020/user/spark/applicationHistory2
spark.eventLog.enabled=true
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)

# 写入数据（首先用R将数据写到硬盘上然后进行装载）
sql(hiveContext, "CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
sql(hiveContext, "LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")

这里需要注意的是：如果运行R的用户不是hdfs管理员用户的话要自己处理一下权限问题，具体问题具体google……

总结

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