集成Hive与Spark SQL及代码实现
问题导读：
1.怎样编译和配置HIVE?
2.怎样配置spark？3.怎样开发spark？
http://www.aboutyun.com/static/image/hrline/4.gif
小结结构上Hive On Spark和SparkSQL都是一个翻译层，把一个SQL翻译成分布式可执行的Spark程序。Hive和SparkSQL都不负责计算。
hive编译如果不是采用CDH在线自动安装和部署的话，可能需要对源码进行编译，使它能够兼容HIVE。
编译的方式也很简单，只需要在Spark_SRC_home（源码的home目录下）执行如下命令：
./make-distribution.sh --tgz -PHadoop-2.2 -Pyarn -DskipTests -Dhadoop.version=2.6.0-cdh5.4.4 -Phive
编译好了之后，会在lib目录下多几个jar包（）
hive配置编辑 $HIVE_HOME/conf/Hive-site.xml,增加如下内容:
<property>
<name>hive.metastore.uris</name>
<value>thrift://master:9083</value>
<description>Thrift uri for the remote metastore. Used by metastore client to connect to remote metastore.</description>
</property>
启动hive metastore
$hive --service metastore &
查看 metastore:
$jobs
+Running hive --service metastore &
关闭 metastore:
$kill %1
kill %jobid,1代表job id
spark配置将 $HIVE_HOME/conf/hive-site.xml copy或者软链 到 $SPARK_HOME/conf/
将 $HIVE_HOME/lib/mysql-connector-java-5.1.12.jar copy或者软链到$SPARK_HOME/lib/
copy或者软链$SPARK_HOME/lib/ 是方便spark standalone模式使用
启动spark-sql1. standalone模式
./bin/spark-sql --master spark:master:7077 --jars /home/spark/spark-1.6.0/lib/mysql-connector-java-5.1.12.jar
2. yarn-client模式
./bin/spark-sql --master yarn-client --jars /home/spark/spark-1.6.0/lib/mysql-connector-java-5.1.12.jar
执行 sql:
select count(*) from people;
3. yarn-cluster模式
Cluster deploy mode 不支持的
./bin/spark-sql --master yarn-cluster--jars /home/spark/spark-1.6.0/lib/mysql-connector-java-5.1.12.jar
Error: Cluster deploy mode is not applicable to Spark SQL shell.
Run with --help for usage help or --verbose for debug output
...
启动 spark-shell1. standalone模式
./bin/spark-shell --master spark:master:7077 --jars /home/spark/spark-1.6.0/lib/mysql-connector-java-5.1.12.jar
2. yarn-client模式
./bin/spark-shell --master yarn-client --jars /home/spark/spark-1.6.0/lib/mysql-connector-java-5.1.12.jar
sqlContext.sql("from people SELECT count(appkey,name1,name2)").collect().foreach(println)
代码测试代码:
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.hive.HiveContext
object SqlHive {
def main(args: Array): Unit = {
val conf=new SparkConf().setAppName("Spark-Hive").setMaster("local")
val sc=new SparkContext(conf)
val hiveContext = new HiveContext(sc)
/*
sqlContext.sql("CREATE TABLE IF NOT EXISTS people (id INT, name STRING, age INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' ")//这里需要注意数据的间隔符
sqlContext.sql("LOAD DATA INPATH '/user/liujiyu/spark/people.txt' INTO TABLE people");
sqlContext.sql(" SELECT * FROM jn1").collect().foreach(println)
*/
//通过HiveContext.table方法来直接加载Hive中的表而生成DataFrame
hiveContext.sql("use hive")
hiveContext.sql("DROP TABLE IF EXISTS people")
hiveContext.sql("CREATE TABLE IF NOT EXISTS people(id INT, name STRING, age INT)")
//当然也可以通过LOAD DATA INPATH去获得HDFS等上面的数据 到Hive（数据的移动）
hiveContext.sql("LOAD DATA LOCAL INPATH '/home/sql/spark/people.txt' INTO TABLE people")
//把本地数据加载到Hive中（数据的拷贝)
hiveContext.sql("DROP TABLE IF EXISTS orders")
hiveContext.sql("CREATE TABLE IF NOT EXISTS orders(peopleId INT, orderNo STRING)")
hiveContext.sql("LOAD DATA LOCAL INPATH '/home/sql/spark/orders.txt' INTO TABLE orders")
//使用join
val resultDF = hiveContext.sql("SELECT p.name, p.age, o.orderNo"
+ "FROM people p JOIN orders o ON p.id=o.peopleId WHERE p.age > 18")
//当删除该表时，磁盘上的数据也会被删除
hiveContext.sql("DROP TABLE IF EXISTS resultT")
//通过saveAsTable的方式把DaraFrame中的数据保存到Hive数据仓库中，数据放在什么地方、元数据都是Hive管理的
resultDF.saveAsTable("resultT")
//通过HivewContext的Table方法去读Hive中的Table并生成DaraFrame
//读取的数据就可以进行机器学习、图计算、各种复杂ETL等操作
val dataFrameHive = hiveContext.table("resultT")
dataFrameHive.show()
sc.stop()
}
}
shell脚本：
./bin/spark-submit --class SparkSQLByScala.SparkSQL2Hive --master spark://slq1:7077 /home/spark/SqlHive.jar
来源：csdn作者：bingo_liu
感谢分享 hello。那个脚本./make-distribution.sh ，我的目录下根本没有啊，是要怎么处理呀？
页:
[1]