hivespark引擎推送数据到es(elasticsearch)

需要导入相关的依赖jar包&＃xff0c;例如es-hadoop的jar包。

这里要特别注意&＃xff1a; 

下载以后要传到hdfs一个地址&＃xff0c;否则无法在hive中添加&＃xff0c;用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&＃61;spark;

创建es外部表&＃xff1a;

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


配置信息&＃xff1a; 

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

每天执行任务先修改别名&＃xff1a; 

alter table profile_v2.shop_user_order SET TBLPROPERTIES (&＃39;es.resource&＃39;&＃61;&＃39;user_profile_${indexdt}&＃39;);

写入数据&＃xff1a;


insert overwrite table profile_v2.shop_user_order
selectshop_id,user_id,all_sale_total_amount,order_cnt,order_quantity_cnt,regexp_replace(first_order_date,&＃39;-&＃39;,&＃39;&＃39;),regexp_replace(last_order_date,&＃39;-&＃39;,&＃39;&＃39;),user_buyback_cycle,order_unit_price,quantity_unit_price,&＃39;shop_user_order&＃39; as hiveTableName,named_struct(&＃39;parent&＃39;,user_id,&＃39;name&＃39;,&＃39;shop_user_order&＃39;)
from profile.ads_shop_all_portrait_shop_user_order
where dt &＃61; &＃39;${dt}&＃39;;


执行脚本&＃xff1a;


dt&＃61;&＃96;date -d &＃39;last day&＃39; &＃43;"%Y%m%d"&＃96;
indexdt&＃61;&＃96;date &＃43;"%Y-%m-%d"&＃96;
echo dt&＃61;$dt ... indexdt&＃61;$indexdtbase_dir&＃61;&＃39;/home&＃39;
sql_name&＃61;&＃39;test.sql&＃39;echo hive --hivevar dt&＃61;${dt} --hivevar indexdt&＃61;${indexdt} -f $base_dir/$sql_name
hive --hivevar dt&＃61;${dt} --hivevar indexdt&＃61;${indexdt} -f $base_dir/$sql_nameif [ $? -ne 0 ];thenecho "hive sql 运行报错&＃xff0c;sql执行失败!"exit -1
fiecho " hive数据导入ES成功 ... "