>>分享SPSS,Hadoop等大数据处理技术,以及分布式架构以及集群系统的构建 书籍支持  卫琴直播  品书摘要  在线测试  资源下载  联系我们
发表一个新主题 开启一个新投票 回复文章 您是本文章第 21660 个阅读者 刷新本主题
 * 贴子主题:  Spark on Yarn with Hive实战案例与常见问题解决 回复文章 点赞(0)  收藏  
作者:flybird    发表时间:2020-01-21 13:37:35     消息  查看  搜索  好友  邮件  复制  引用

    

1 场景

在实际过程中,遇到这样的场景:
  日志数据打到HDFS中,运维人员将HDFS的数据做ETL之后加载到hive中,之后需要使用Spark来对日志做分析处理,Spark的部署方式是Spark on Yarn的方式。
从场景来看,需要在我们的Spark程序中通过HiveContext来加载hive中的数据。

     如果希望自己做测试,环境的配置可以参考我之前的文章,主要有下面的需要配置:
  • 1.Hadoop环境
    • Hadoop环境的配置可以参考之前写的文章;
  • 2.Spark环境
    • Spark环境只需要在提交job的节点上进行配置即可,因为使用的是Spark on Yarn的方式;
  • 3.Hive环境
    • 需要配置好Hive环境,因为在提交Spark任务时,需要连同hive-site.xml文件一起提交,因为只有这样才能够识别已有的hive环境的元数据信息;
    • 所以其实中Spark on Yarn的部署模式中,需要的只是hive的配置文件,以让HiveContext能够读取存储在mysql中的元数据信息以及存储在HDFS上的hive表数据;
    • hive环境的配置可以参考之前的文章;
其实之前已经有写过Spark Standalone with Hive的文章,可以参考:《Spark SQL笔记整理(三):加载保存功能与Spark SQL函数》

2 编写程序与打包

作为一个测试案例,这里的测试代码比较简单,如下:    

package cn.xpleaf.spark.scala.sql.p2

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @author xpleaf
  */

object _01HiveContextOps {

    def main(args: Array[String]): Unit = {
        Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
        val conf = new SparkConf()
//            .setMaster("local[2]")
            .setAppName(s"${_01HiveContextOps.getClass.getSimpleName}")

        val sc = new SparkContext(conf)
        val hiveContext = new HiveContext(sc)

        hiveContext.sql("show databases").show()

        hiveContext.sql("use mydb1")
        // 创建teacher_info表
        val sql1 = "create table teacher_info(
" + "name string,
" + "height double)
" + "row format delimited
" + "fields terminated by ','"
        hiveContext.sql(sql1)

        // 创建teacher_basic表
        val sql2 = "create table teacher_basic(
" + "name string,
" + "age int,
" + "married boolean,
" + "children int)
" + "row format delimited
" + "fields terminated by ','"
        hiveContext.sql(sql2)

        // 向表中加载数据
        hiveContext.sql("load data inpath 'hdfs://ns1/data/hive/teacher_info.txt' into table teacher_info")
        hiveContext.sql("load data inpath 'hdfs://ns1/data/hive/teacher_basic.txt' into table teacher_basic")

        // 第二步操作:计算两张表的关联数据
        val sql3 = "select
" + "b.name,
" + "b.age,
" + "if(b.married,'已婚','未婚') as married,
" + "b.children,
" + "i.height
" + "from teacher_info i
" + "inner join teacher_basic b on i.name=b.name"
        val joinDF:DataFrame = hiveContext.sql(sql3)

        val joinRDD = joinDF.rdd
        joinRDD.collect().foreach(println)

        joinDF.write.saveAsTable("teacher")

        sc.stop()
    }

}

  可以看到其实只是简单的在hive中建表、加载数据、关联数据与保存数据到hive表中。

     编写完成之后打包就可以了,注意不需要将依赖一起打包。之后就可以把jar包上传到我们的环境中了。

3 部署

编写submit脚本,如下:    

[hadoop@hadoop01 jars]$ cat spark-submit-yarn.sh
/home/hadoop/app/spark/bin/spark-submit \
--class $2 \
--master yarn \
--deploy-mode cluster \
--executor-memory 1G \
--num-executors 1 \
--files $SPARK_HOME/conf/hive-site.xml \
--jars $SPARK_HOME/lib/mysql-connector-java-5.1.39.jar,$SPARK_HOME/lib/datanucleus-api-jdo-3.2.6.jar,$SPARK_HOME/lib/datanucleus-core-3.2.10.jar,$SPARK_HOME/lib/datanucleus-rdbms-3.2.9.jar \
$1 \

  注意其中非常关键的

--files



--jars

,说明如下:    

--files $HIVE_HOME/conf/hive-site.xml    //将Hive的配置文件添加到Driver和Executor的classpath中
--jars $HIVE_HOME/lib/mysql-connector-java-5.1.39.jar,….    //将Hive依赖的jar包添加到Driver和Executor的classpath中

  之后就可以执行脚本,将任务提交到Yarn上:    

[hadoop@hadoop01 jars]$ ./spark-submit-yarn.sh spark-process-1.0-SNAPSHOT.jar cn.xpleaf.spark.scala.sql.p2._01HiveContextOps

4 查看结果

  需要说明的是,如果需要对执行过程进行监控,就需要进行配置historyServer(mr的jobHistoryServer和spark的historyServer),可以参考我之前写的文章。

4.1 Yarn UI

点击在新窗口中浏览原图
CTRL+鼠标滚轮放大或缩小

     点击在新窗口中浏览原图
CTRL+鼠标滚轮放大或缩小

4.2 Spark UI

点击在新窗口中浏览原图
CTRL+鼠标滚轮放大或缩小

     点击在新窗口中浏览原图
CTRL+鼠标滚轮放大或缩小

4.3 Hive

可以启动hive,然后查看我们的spark程序加载的数据:    

hive (mydb1)>
            >
            >
            > show tables;
OK
t1
t2
t3_arr
t4_map
t5_struct
t6_emp
t7_external
t8_partition
t8_partition_1
t8_partition_copy
t9
t9_bucket
teacher
teacher_basic
teacher_info
test
tid
Time taken: 0.057 seconds, Fetched: 17 row(s)
hive (mydb1)> select *
            > from teacher_info;
OK
zhangsan        175.0
lisi    180.0
wangwu  175.0
zhaoliu 195.0
zhouqi  165.0
weiba   185.0
Time taken: 1.717 seconds, Fetched: 6 row(s)
hive (mydb1)> select *
            > from teacher_basic;
OK
zhangsan        23      false   0
lisi    24      false   0
wangwu  25      false   0
zhaoliu 26      true    1
zhouqi  27      true    2
weiba   28      true    3
Time taken: 0.115 seconds, Fetched: 6 row(s)
hive (mydb1)> select *
            > from teacher;
OK
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
zhangsan        23      未婚    0       175.0
lisi    24      未婚    0       180.0
wangwu  25      未婚    0       175.0
zhaoliu 26      已婚    1       195.0
zhouqi  27      已婚    2       165.0
weiba   28      已婚    3       185.0
Time taken: 0.134 seconds, Fetched: 6 row(s)

5 问题与解决

  1.User class threw exception: java.lang.RuntimeException: java.lang.RuntimeException: Unable to instantiate org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient
注意我们的Spark部署模式是Yarn,yarn上面是没有相关spark和hive的相关依赖的,所以在提交任务时,必须要指定要上传的jar包依赖:    

--jars $SPARK_HOME/lib/mysql-connector-java-5.1.39.jar,$SPARK_HOME/lib/datanucleus-api-jdo-3.2.6.jar,$SPARK_HOME/lib/datanucleus-core-3.2.10.jar,$SPARK_HOME/lib/datanucleus-rdbms-3.2.9.jar \

  其实在提交任务时,注意观察控制台的输出:    

18/10/09 10:57:44 INFO yarn.Client: Uploading resource file:/home/hadoop/app/spark/lib/spark-assembly-1.6.2-hadoop2.6.0.jar -> hdfs://ns1/user/hadoop/.sparkStaging/application_1538989570769_0023/spark-assembly-1.6.2-hadoop2.6.0.jar
18/10/09 10:57:47 INFO yarn.Client: Uploading resource file:/home/hadoop/jars/spark-process-1.0-SNAPSHOT.jar -> hdfs://ns1/user/hadoop/.sparkStaging/application_1538989570769_0023/spark-process-1.0-SNAPSHOT.jar
18/10/09 10:57:47 INFO yarn.Client: Uploading resource file:/home/hadoop/app/spark/lib/mysql-connector-java-5.1.39.jar -> hdfs://ns1/user/hadoop/.sparkStaging/application_1538989570769_0023/mysql-connector-java-5.1.39.jar
18/10/09 10:57:47 INFO yarn.Client: Uploading resource file:/home/hadoop/app/spark/lib/datanucleus-api-jdo-3.2.6.jar -> hdfs://ns1/user/hadoop/.sparkStaging/application_1538989570769_0023/datanucleus-api-jdo-3.2.6.jar
18/10/09 10:57:47 INFO yarn.Client: Uploading resource file:/home/hadoop/app/spark/lib/datanucleus-core-3.2.10.jar -> hdfs://ns1/user/hadoop/.sparkStaging/application_1538989570769_0023/datanucleus-core-3.2.10.jar
18/10/09 10:57:47 INFO yarn.Client: Uploading resource file:/home/hadoop/app/spark/lib/datanucleus-rdbms-3.2.9.jar -> hdfs://ns1/user/hadoop/.sparkStaging/application_1538989570769_0023/datanucleus-rdbms-3.2.9.jar
18/10/09 10:57:47 INFO yarn.Client: Uploading resource file:/home/hadoop/app/spark/conf/hive-site.xml -> hdfs://ns1/user/hadoop/.sparkStaging/application_1538989570769_0023/hive-site.xml
18/10/09 10:57:47 INFO yarn.Client: Uploading resource file:/tmp/spark-6f582e5c-3eef-4646-b8c7-0719877434d8/__spark_conf__103916311924336720.zip -> hdfs://ns1/user/hadoop/.sparkStaging/application_1538989570769_0023/__spark_conf__103916311924336720.zip

  也可以看到,其会将相关spark相关的jar包上传到yarn的环境也就是hdfs上,之后再执行相关的任务。
  2.User class threw exception: org.apache.spark.sql.execution.QueryExecutionException: FAILED: SemanticException [Error 10072]: Database does not exist: mydb1
mydb1不存在,说明没有读取到我们已有的hive环境的元数据信息,那是因为在提交任务时没有指定把hive-site.xml配置文件一并提交,如下:    

--files $SPARK_HOME/conf/hive-site.xml \

----------------------------
原文链接:https://blog.51cto.com/xpleaf/2296151

程序猿的技术大观园:www.javathinker.net



[这个贴子最后由 flybird 在 2020-01-21 20:53:20 重新编辑]
  Java面向对象编程-->图形用户界面(上)
  JavaWeb开发-->Servlet技术详解(Ⅱ)
  JSP与Hibernate开发-->映射一对多关联关系
  Java网络编程-->通过JavaMail API收发邮件
  精通Spring-->Vue Router路由管理器
  Vue3开发-->Vue简介
  springboot的缓存技术
  Spark批量读取Redis数据-Pipeline(Scala)
  mongodb与redis与Hbase比较
  kafka+spark-streaming实时推荐系统性能优化笔记
  如何面对高并发?缓存?中台为什么会火?
  搭建高可用的Replication集群归档大量的冷数据
  Spark高级排序与TopN问题揭密
  Spark On YARN 环境搭建
  Hadoop2.6.5+centos7.5三节点大数据集群部署搭建
  hive 的支持的文件类型与 压缩格式
  Zookeeper+Kafka集群搭建
  Hadoop的简单单词统计案例
  Hadoop起源及其四大特性详解
  深入剖析Hadoop HBase
  MapReduce工作原理图文详解
  更多...
 IPIP: 已设置保密
树形列表:   
1页 1条记录 当前第1
发表一个新主题 开启一个新投票 回复文章


中文版权所有: JavaThinker技术网站 Copyright 2016-2026 沪ICP备16029593号-2
荟萃Java程序员智慧的结晶,分享交流Java前沿技术。  联系我们
如有技术文章涉及侵权,请与本站管理员联系。