Hive编程指南-Spark操作Hive

本是古典何须时尚 2021-09-11 07:00 674阅读 0赞

摘要：本文将要说明如何使用Spark来对Hive进行操作

# 1、打jar包，提交Spark任务 #

通过提交spark任务的方式，如下面的scala代码。之后需要将成代码打包成一个jar包，然后提交到spark中去.一般情况下生产上建议使用这种方法，可以灵活控制sql里的各项参数

val conf = new SparkConf()
        val sc = new SparkContext(conf)
        val hiveContext = new HiveContext(sc)
        hiveContext.setConf("hive.support.concurrency", "true")
        hiveContext.setConf("hive.exec.parallel", "true")
        hiveContext.setConf("hive.exec.dynamic.partition", "true")
        hiveContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
        hiveContext.setConf("hive.exec.max.dynamic.partitions.pernode", "10000")
        hiveContext.sql(s"insert overwrite table $mtagoPblIdxd_table partition(statis_time)  select  mtagoPbl,pv,uv,uvNewVisitor,uvRegUcnt,uvCrdUcnt,uvActUcnt,regUcnt,crdApplyUcnt,crdPassUcnt,crdAmt,bindCardUcnt,bowCnt,bowUcnt,bowAmt,ordCnt,ordUcnt,ordAmt,ordDiscountAmt,ordPayCnt,ordPayUcnt,ordPayAmt,ordUsecrdAmt,loanUcnt,repayCnt,repayUcnt,repayAmt,createTime,updateDate,updateTime,statisTime from $temp_MtagoPblIdxd_table")

之后就是使用命令来提交任务

./bin/spark-submit --master  。。。 --class 。。。--executor-memory 。。。

好处：

(1)可以结合rdd来一起使用，比如使用这样写：

val result1 = hivecontext.sql("select \* from ...").rdd.filter.reduceBykey().....

(2)可以传递不用的参数，如

hiveContext.sql(s"insert overwrite table $mtagoPblIdxd_。。。）这种参数。可以动态注入

(3)方便调试，代码也比较容易组织

缺点：前期开发准备工作要花费一定时间。

# 2、使用spark-shell命令行 #

使用方法如下：

先进入spark交互界面

![20170215204220823][]

默认情况会启动一下sqlcontext,创建一个hivecontext

![20170215204307559][]

然后就可以使用hiveContext来提交sql了！

hiveContext.sql("select chnl_code chnl_code, count(1)  bow_cnt,sum(case when bow_amt is null then 0 else bow_amt end)  bow_amt from ods_uba.bus_req where req_date = '2016-10-23' and bus_type = 'bow' group by chnl_code").rdd.collect().foreach(println);

优点：

上手快，能很快验证sql的准确性或查询一些数据

缺点：

不好将数据进行进一步的处理，或者存储到其它地方

# 3、通过spark-sql #

spark是也提供了直接调用shell脚本的功能，

编写如下clean.sh脚本

#!/bin/sh  
    /app/hadoop/spark/bin/spark-sql --master spark://sach01-005:7077  -i //app/hadoop/linlin.sql - --executor-memory 1g -e "\
    insert overwrite table bus_req_prc partition (req_date, bus_type, bus_subtype) 
       select cr.req_no, cr.user_id, cr.mbl_nbr, cr.uuid_type, cr.uuid, cr.openid_type, cr.openid, cr.chnl_code, 
             cr.mtago_p, cr.mtago_pb, cr.mtago_pbl,
             cr.mtagi_p, cr.mtagi_pb, cr.mtagi_pbl,
             cr.mtagc_p, cr.mtagc_pb, cr.mtagc_pbl,
             cr.fr_page_type, cr.fr_page_id, cr.fr_evt_type, cr.fr_evt_id,
             cr.req_time, rrs.customer_no,
    
             null, null,    -- register
             null, null, null,  -- bind card
    
             null, null, null, null, null, null, null, 0, -- credit
    
             null, 0, 0, 0, null, null, -- borrow    
    
             null, 0, 0, null, null, -- ebp order
    
             0, 0, 0, 0, 0, null, null,  -- loan
             
             rrs.application_form_id, 
             rrs.actual_repay_amount, 
             rrs.req_date, 
             substr(rrs.request_date, 11), 
    
             from_unixtime(unix_timestamp(), 'yyyy-MM-dd HH:mm:ss'),
             from_unixtime(unix_timestamp(), 'yyyy-MM-dd'),
             from_unixtime(unix_timestamp(), 'HH:mm:ss'),
    
             '2016-17-12', 'repay', 'rbs'
             from repayment_request_summary as rrs
             join  chnl_req as cr on (cr.req_no = rrs.trans_ref_no)
             where rrs.req_date='2016-17-12';
      "\

然后执行sh clean.sh即可

\-i 表示执行指定后缀为.sql的脚本

\-e 表示执行后面的内容

优点：

可以直接将原本用Hive执行MapReduce执行的shell脚本，sql脚本基本不用做太大的改动就可以使用spark来跑批（根据我们使用的经验，速度提交至少都是2倍以上）

缺点：

还是无法将数据进行进一步的加工处理，如何使用rdd的函数等，并且维护的成本也不低。

[20170215204220823]: /images/20210811/843181b0b3da4c3e9fe24e11719a5e90.png
[20170215204307559]: /images/20210811/1ae8181381e44b648d39efb84a68de36.png