Spark Sql ON Hive

一、配置文件

1. spark-defaults.conf

  • 详见配置文件

2. hive 集成配置

  1. 1. hive-site.xml
  3. <configuration>
  4.   <property>
  5.     <name>hive.metastore.uris</name>
  6.     <value>thrift://dw1:9083,thrift://dw2:9083</value>
  7.   </property>
  8.   <property>
  9.     <name>hive.metastore.client.socket.timeout</name>
  10.     <value>300</value>
  11.   </property>
  12.   <property>
  13.     <name>hive.metastore.warehouse.dir</name>
  14.     <value>/user/hive/warehouse</value>
  15.   </property>
  16. </configuration>
  19. 2. spark-env.sh
  20.   export HADOOP_HOME=/usr/lib/hadoop
  21.   export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
  22.   export SPARK_LIBRARY_PATH=$SPARK_LIBRARY_PATH:$HADOOP_HOME/lib/native:$HADOOP_HOME/lib

二、SparkSQL 客户端模式部署

  1. -- 普通加载模式
  2. spark-sql \
  3. --master yarn \
  4. --deploy-mode client \
  5. --name spark-sql-service_1 \
  6. --driver-cores 2 \
  7. --driver-memory 4096M \
  8. --executor-cores 1 \
  9. --executor-memory 2048M \
  10. --num-executors 3 \
  11. --jars file:///etc/hive/auxlib/dw_hive_udf-1.0.jar,file:///etc/hive/auxlib/json-serde-1.3.7-jar-with-dependencies.jar
  14. -- 配置参数加载模式
  15. spark-sql \
  16. --master yarn \
  17. --deploy-mode client \
  18. --name spark-sql-service \
  19. --driver-cores 2 \
  20. --driver-memory 4096M \
  21. --executor-cores 1 \
  22. --executor-memory 2048M \
  23. --num-executors 3 \
  24. --conf spark.driver.extraJavaOptions="-DJAVA_LIBRARY_PATH=/opt/cloudera/parcels/CDH/lib/hadoop/lib/native:$LD_LIBRARY_PATH" \
  25. --jars file:///etc/hive/auxlib/dw_hive_udf-1.0.jar,file:///etc/hive/auxlib/json-serde-1.3.7-jar-with-dependencies.jar

三、SparkSQL ThriftServer 模式部署

1. 启动 Thriftserver 进程

  • 更多配置见 spark-defaults.conf
  2. 1. Yarn 模式
  3.   (1) yarn-client 模式 , 让 Yarn 管理进程 , Driver 运行在客户端 ,Work 运行在 NodeManager 上
  4.     $SPARK_HOME/sbin/start-thriftserver.sh \
  5.     --master yarn \
  6.     --deploy-mode client \
  7.     --name spark-sql-service \
  8.     --queue root.default \
  9.     --driver-cores 2 \
  10.     --driver-memory 4096M \
  11.     --executor-cores 1 \
  12.     --executor-memory 2048M \
  13.     --num-executors 3 \
  14.     --jars file://path/xxx.jar,file://path/xxx.jar \
  15.     --hiveconf hive.server2.thrift.port=10010
  17.   (2) yarn-cluster模式, 集群模式目前不支持
  18.     ./sbin/start-thriftserver.sh \
  19.     --master yarn \
  20.     --deploy-mode client \
  21.     --name spark-sql-service \
  22.     --driver-cores 2 \
  23.     --driver-memory 4096M \
  24.     --executor-cores 1 \
  25.     --executor-memory 2048M \
  26.     --num-executors 3 \
  27.     --hiveconf hive.server2.thrift.port=10010
  29. 2. standalone 模式
  30.   $SPARK_HOME/sbin/start-thriftserver.sh \
  31.   --master spark://uhadoop-ociicy-task3:7077 \
  32.   --deploy-mode client \
  33.   --name spark-sql \
  34.   --driver-cores 2 \
  35.   --driver-memory 500M \
  36.   --hiveconf hive.server2.thrift.port=10010
  39. 3. JDBC 操作 hive
  40.   $SPARK_HOME/bin/beeline !connect jdbc:hive2://hostname:10010

2. SparkSQL thrift-server HA 模式

2.1 配置参数

  1. <property>
  2.   <name>hive.server2.support.dynamic.service.discovery</name>
  3.   <value>true</value>
  4. </property>
  5. <property>
  6.   <name>hive.server2.zookeeper.namespace</name>
  7.   <value>sparkserver2_zk</value>
  8. </property>
  9. <property>
  10.   <name>hive.zookeeper.quorum</name>
  11.   <value>master1:2181,master2:2181,node1:2181</value>
  12. </property>
  13. <property>
  14.   <name>hive.zookeeper.client.port</name>
  15.   <value>2181</value>
  16. </property>
  18. <!-- HiveServer2 启动的节点地址(根据 HA 的部署节点的名称写, 根据部署节点配置) -->
  19. <property>
  20.   <name>hive.server2.thrift.bind.host</name>
  21.   <value>node9</value>
  22. </property>

2.2 部署服务

  1. 1. Spark 编译支持 HA 模式的 Jar 包
  3.   spark-hive-thriftserver_2.11-2.0.2.jar 加入到 $SPARK_HOME/jars/spark-hive-thriftserver_2.11-2.0.2.jar
  6. 2. 部署配置, 这里需要在 2 个节点, 同时启动服务
  8.   $SPARK_HOME/sbin
  9.   ./start-thriftserver.sh \
  10.   --master yarn \
  11.   --name service_name \
  12.   --conf spark.driver.memory=3G \
  13.   --executor-memory 1G \
  14.   --num-executors 10 \
  15.   --hiveconf hive.server2.thrift.port=10003
  18. 3. 测试连接 HA 模式下的 SparkThriftService
  20. $SPARK_HOME/bin/beeline -u "jdbc:hive2://zkNode1:2181,zkNode2:2181,zkNode3:2181/default;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=sparkserver2_zk" hadoop hadoop
  22. # 客户端方式启动
  23. $HIVE_HOME/bin/beeline
  24. > !connect jdbc:hive2://zkNode1:2181,zkNode2:2181,zkNode3:2181/default;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=sparkserver2_zk hadoop hadoop

3. Spark 动态资源部署

  • 具体参数说明见: spark-defaults.conf
  1. 1. 配置 yarn-site.xml 支持动态资源部署模式( NodeManager 所有节点上, 重启服务)
  3. ##### NodeManager 附属程序支持 Start #####
  5. <!-- NodeManager 上运行的附属服务, 用于提升 Shuffle 计算性能 -->
  6. <property>
  7.   <name>yarn.nodemanager.aux-services</name>
  8.   <value>mapreduce_shuffle,spark_shuffle</value>
  9. </property>
  11. <!-- NodeManager 中辅助服务对应的类 -->
  12. <property>
  13.   <name>yarn.nodemanager.aux-services.spark_shuffle.class</name>
  14.   <value>org.apache.spark.network.yarn.YarnShuffleService</value>
  15. </property>
  17. <!-- Shuffle 服务监听数据获取请求的端口。可选配置, 默认值为 7337 -->
  18. <property>
  19.   <name>spark.shuffle.service.port</name>
  20.   <value>7337</value>
  21. </property>
  23. ##### NodeManager 附属程序支持 End #####
  26. 2. 配置 spark-defaults.conf 配置动态分配
  28. ##### 动态分配 Start #####
  30. ## 注意事项：根据任务动态向 yarn 申请资源, 会导致申请资源浪费大量时间。
  32. ## 是否使用动态资源分配，它根据工作负载调整为此应用程序注册的执行程序数量。
  33. spark.dynamicAllocation.enabled                       true
  34. ## 使用外部 shuffle, 保存了由 executor 写出的 shuffle 文件所以 executor 可以被安全移除, spark.dynamicAllocation.enabled 为 true, 这个选项才可以为 true
  35. spark.shuffle.service.enabled                         true
  37. ## 每个 Application 最小分配的 executor 数
  38. spark.dynamicAllocation.minExecutors                  3
  39. ## 每个 Application 最大并发分配的 executor 数。 ThriftServer 模式是整个 ThriftServer 同时并发的最大资源数，如果多个用户同时连接，则会被多个用户共享竞争
  40. spark.dynamicAllocation.maxExecutors                  6
  42. ## 如果启用动态分配，并且有超过此持续时间的挂起任务积压，则将请求新的执行者。
  43. spark.dynamicAllocation.schedulerBacklogTimeout       6s
  44. ## 与 spark.dynamicAllocation.schedulerBacklogTimeout 相同，但仅用于后续执行者请求
  45. spark.dynamicAllocation.sustainedSchedulerBacklogTimeout              6s
  47. ## 如果启用动态分配，并且执行程序已空闲超过此持续时间，则将删除执行程序。
  48. spark.dynamicAllocation.executorIdleTimeout           60s
  50. ##### 动态分配 End #####
  53. 2. 部署(spark 默认会读取 spark-defaults.conf 配置文件, 但是命令行启动会覆盖 spark-defaults.conf)
  55. $SPARK_HOME/sbin/start-thriftserver.sh \
  56. --master yarn \
  57. --deploy-mode client \
  58. --queue root.default \
  59. --name test \
  60. --driver-cores 4 \
  61. --driver-memory 8192M \
  62. --executor-cores 6 \
  63. --executor-memory 12288M \
  64. --conf spark.dynamicAllocation.enabled=true \
  65. --conf spark.dynamicAllocation.minExecutors=3 \
  66. --conf spark.dynamicAllocation.maxExecutors=6 \
  67. --hiveconf hive.server2.thrift.port=10010 \
  68. --jars file:///etc/hive/auxlib/dw_hive_udf-1.0.jar,file:///etc/hive/auxlib/json-serde-1.3.7-jar-with-dependencies.jar

三、Spark Sql Udf

  • Hive UDF 与 Spark UDF 通用
  • UDF

四、Spark Sql 编程

  1. package com.dw2345.machine_learn.combination.sql;
  3. // $example on:spark_hive$
  4. import org.apache.spark.sql.Row;
  5. import org.apache.spark.sql.SparkSession;
  6. // $example off:spark_hive$
  8. object TestSparkSql {
  9.     def main(args: Array[String]) {
  10.         // warehouseLocation points to the default location for managed databases and tables
  11.         var warehouseLocation = "hdfs://nameservice1/user/hive/warehouse"
  13.         // 需要 hive-site.xml 文件和 hdfs-site.xml 文件
  14.         val spark = SparkSession
  15.           .builder()
  16.           .master("local")
  17.           .appName("Spark Hive Example")
  18.           .config("spark.sql.warehouse.dir", warehouseLocation)
  19.           .enableHiveSupport()
  20.           .getOrCreate()
  22.         import spark.implicits._
  23.         import spark.sql
  25.         // Queries are expressed in HiveQL
  26.         sql("SELECT * FROM web_logs_text LIMIT 10").show()
  27.         // sql("SELECT * FROM dm_db.dm_channel_inst_compete WHERE p_dt='2017-06-30' AND p_hours='11' LIMIT 10").show()
  28.         // sql("SELECT * FROM ods.ods_pic_use WHERE p_type='2' AND p_dt='2017-08-28' AND p_hours='23' LIMIT 10").show()
  30.     }
  31. }