hive spark引擎推送数据到es(elasticsearch)

需要导入相关的依赖jar包，例如es-hadoop的jar包。

这里要特别注意： 

下载以后要传到hdfs一个地址，否则无法在hive中添加，用hdfs dfs -put命令。

--导入依赖jar包
add jar hdfs://elasticsearch-hadoop-7.13.0.jar;
add jar hdfs://commons-httpclient-3.1.jar;
--设置hive引擎为spark引擎
set hive.execution.engine=spark;

创建es外部表：

create EXTERNAL TABLE IF NOT EXISTS profile_v2.shop_user_order(
  shop_id                    string     comment'店铺id'
  ,user_id                    string     comment'会员id'
  ,all_sale_total_amount      bigint     comment'累计下单金额(2016年开始支付的不考虑退货)单位：分'
  ,order_cnt                  bigint     comment'下单次数(2016年开始支付的)'
  ,order_quantity_cnt         bigint     comment'下单件数(2016年开始支付的)'
  ,first_order_date           string     comment'首次下单时间(2016年开始支付yyyymmdd)'
  ,last_order_date            string     comment'最后下单时间(2016年开始支付yyyymmdd)'
  ,user_buyback_cycle         bigint     comment'客户回购周期(最后-首次)/下单次数)'
  ,order_unit_price           bigint     comment'客单价(累计下单金额/下单次数)'
  ,quantity_unit_price        bigint     comment'件单价(累计下单金额/下单件数)'
  ,hivetablename              string
  ,join_field                 struct
)
comment '店铺-会员-历史下单行为分析(全量)'
--指定存储格式为es相关
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES(
  'es.resource' = 'user_profile',
  'es.nodes'='10.112',
  'es.port'='0',
  'es.index.auto.create'='TRUE',
  --'es.mapping.id' = 'user_id',
  'es.nodes.wan.only'='true',
  'es.read.metadata'='true',
  'es.mapping.names'='hivetablename:hiveTableName',
  'es.mapping.routing'='user_id',
  'es.mapping.join'='user_id'
);

配置信息： 

 --指定索引
  'es.resource' = 'user_profile',
  --es节点 多个用,分隔
  'es.nodes'='10.112',
  --es端口
  'es.port'='80',
  --是否自动创建索引
  'es.index.auto.create'='TRUE',
  --表示es的_id对应的字段
  --'es.mapping.id' = 'user_id',
  --默认为 false，设置为 true 之后，会关闭节点的自动 discovery，只使用es.nodes声明的节点进行数据读写操作；如果你需要通过域名进行数据访问，则设置该选项为 true，否则请务必设置为 false；
  'es.nodes.wan.only'='true',
  --要想把_id映射到Hive表字段中，必须使用这种方式
  'es.read.metadata'='true',
  --表示其他字段的对应(可以不写，插入时按顺序依次对应即可);
  'es.mapping.names'='hivetablename:hiveTableName',
  'es.mapping.routing'='user_id',
   'es.mapping.join'='user_id'

每天执行任务先修改别名： 

alter table profile_v2.shop_user_order SET TBLPROPERTIES ('es.resource'='user_profile_${indexdt}');

写入数据：

insert overwrite table  profile_v2.shop_user_order
select
  shop_id
  ,user_id
  ,all_sale_total_amount
  ,order_cnt
  ,order_quantity_cnt
  ,regexp_replace(first_order_date,'-','')
  ,regexp_replace(last_order_date,'-','')
  ,user_buyback_cycle
  ,order_unit_price
  ,quantity_unit_price
  ,'shop_user_order' as hiveTableName
  ,named_struct('parent',user_id,'name','shop_user_order')
from profile.ads_shop_all_portrait_shop_user_order
where dt = '${dt}';

执行脚本：

dt=`date -d 'last day' +"%Y%m%d"`
indexdt=`date +"%Y-%m-%d"`
echo dt=$dt ... indexdt=$indexdt

base_dir='/home'
sql_name='test.sql'

echo hive --hivevar dt=${dt} --hivevar indexdt=${indexdt} -f  $base_dir/$sql_name
hive --hivevar dt=${dt} --hivevar indexdt=${indexdt} -f  $base_dir/$sql_name

if [ $? -ne 0 ];then
    echo "hive sql 运行报错，sql执行失败!"
    exit -1
fi

echo " hive数据导入ES成功 ... "