Hive

1.现象 2.原因 集群资源使用率过高时可能会导致Hive On Spark查询失败-查询超时。 从hive on spark的架构看出超时的位置： 3.解决 修改以下参数，重启集群 ### 其他可设置的参考参数 # 在Hive client和远程Spark driver通信过程中，随机生成密码的比特数。最好设置成8的倍数。　hive.spark.client.secret.bits # 远程Spark drive用于处理RPC事件所用的最大线程数，默认是8。　hive.spark.client.rpc.threads # Hive client和远程Spark driver通信最大的消息大小（单位：byte）,默认是50MB。　hive.spark.client.rpc.max.size # 远程Spark driver的通道日志级别，必须是DEBUG, ERROR, INFO, TRACE, WARN中的一个。　hive.spark.client.channel.log.level # 用于身份验证的SASL机制的名称。 hive.spark.client.rpc.sasl.mechanisms #生产集群设置的相应参数： hive.spark.client.future.timeout=360s # Hive client请求Spark driver的超时时间，如果没有指定时间单位，默认就是秒。 hive.metastore.client.socket.timeout=360s # 客户端socket超时时间，默认20秒。 hive.spark.client.connect.timeout=360000ms # Spark driver连接Hive client的超时时间，如果没有指定时间单位，默认就是毫秒。 hive.spark.client.server.connect.timeout=360000ms # Hive client和远程Spark driver握手时的超时时间，这个会在两边都检查的，如果没有指定时间单位，默认就是毫秒。 hive.spark.job.monitor.timeout=180s # Job监控获取Spark作业状态的超时时间，如果没有指定时间单位，默认就是秒。

在es中数据类型为date： "addTime": { "format": "yyyy-MM-dd HH:mm:ss", "type": "date" } 在hive建映射表 CREATE EXTERNAL TABLE hive_es.cty_test1( addTime date ) STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler' TBLPROPERTIES('es.resource' = 'cty_test/cty_test', 'es.nodes'='172.16.98.113,172.16.98.149,172.16.98.150,172.16.98.151,172.16.98.152', 'es.port'='9200', 'es.mapping.names'= 'addTime:addTime', 'es.date.format'='yyyy-MM-dd HH:mm:ss', 'es.index.auto.create'='false', ) 查询报错： 更改hive表数据类型为string CREATE EXTERNAL TABLE hive_es.cty_test5( addTime string ) STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler' TBLPROPERTIES('es.resource' = 'cty_test/cty_test', 'es.nodes'='172.16.98.113,172.16.98.149,172.16.98.150,172.16.98.151,172.16.98.152', 'es.port'='9200', 'es.mapping.names'= 'addTime:addTime', 'es.date.format'='yyyy-MM-dd HH:mm:ss', 'es.index.auto.create'='false', ) 查询继续报错： 查阅资料： elasticsearch-hadoop中用于将ES中的日期转换为Hive中的日期格式的类为org.elasticsearch.hadoop.hive.HiveValueReader，通过查看该类的源码，其实现的用户日期转换的方法为： @Override protected Object parseDate(String value, boolean richDate) { return (richDate ? new TimestampWritable(new Timestamp(DatatypeConverter.parseDateTime(value).getTimeInMillis())) : parseString(value)); } 可以看到它是通过javax.xml.bind.DatatypeConverter.parseDateTime(String)方法将对应的日期字符串转换为日期的，该方法不支持的日期字符串格式为“yyyy-MM-dd HH:mm:ss”的字符串，它支持的日期字符串的格式为“yyyy-MM-ddTHH:mm:ss”这样的。 解决方案： 在建表时设置参数’es.mapping.date.rich’=‘false’，然后hive字段类型设为string。 官方解释： Whether to create a rich Date like object for Date fields in Elasticsearch or returned them as primitives (String or long). By default this is true. The actual object type is based on the library used; noteable exception being Map/Reduce which provides no built-in Date object and as such LongWritable and Text are returned regardless of this setting. ...

zzz1 uid is_delete 1 1 2 0 3 0 zzz2 uid is_delete 1 0 2 1 3 0 select * from zzz1 left join zzz2 on zzz1.uid = zzz2.uid where zzz1.is_delete = 0 and zzz2.is_delete = 0 结果: 3 0 3 0 select * from (select * from zzz1 where zzz1.is_delete = 0) z1 left join (select * from zzz2 where zzz2.is_delete = 0) z2 on z1.uid = z2.uid 结果: 2 0 NULL NULL 3 0 3 0 select * from (select * from zzz1 where zzz1.is_delete = 0) z1 left join (select * from zzz2 where zzz2.is_delete = 0) z2 on z1.uid = z2.uid where z2.uid is not null 结果: 3 0 3 0 ...

1.创建临时函数 hive> add jar /home/hadoop/bigdata_udf.jar; hive> create temporary function isContains100 as 'com.xx.hive.udf.hm2.IsContains100'; --验证 hive> select isContains100(t.col1) from t limit 10; hive> drop temporary function isContains100; 2.创建永久函数 hadoop fs -put /opt/bigdata_udf.jar /udf hive> create function default.url_decode as 'com.xx.udf.DecodeURL' using jar 'hdfs:///udf/bigdata_udf.jar'; --验证 hive> select default.url_decode(t.col1) from t limit 10; 注意:注册永久函数必须使用hdfs路径,不可使用本地路径

CREATE EXTERNAL TABLE `mongodb_dingtalk.hbasetohive_patent`( key String, abstracts String, address String) STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,info:abstracts,info:address") TBLPROPERTIES("[hbase.table.name](http://hbase.table.name/)" = "dingtalk:patent"); 仅从HBase拉取数据使用，禁止利用此种方式往HBase写数据 数据拉出时拉出hbase中数据对应当前版本的时间戳 create external table ods.ods_zxk_hbase_wechat_public_account_mapping( id string comment '需要拆分的key', ts timestamp comment '数据标识符，用来跟新数据做去重') comment '高级搜索有无微信公众号' STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' WITH SERDEPROPERTIES ('hbase.columns.mapping' = ':key,:timestamp') TBLPROPERTIES('[hbase.table.name](http://hbase.table.name/)' = 'dingtalk:wechat_public_account')

一 hive导入es 1 创建hive-es映射表 CREATE EXTERNAL TABLE hive_es.re_run_test2( id STRING ,test STRING) STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler' TBLPROPERTIES('es.resource' = 're_run_test2/test', 'es.nodes'='172.16.98.113,172.16.98.149,172.16.98.150,172.16.98.151,172.16.98.152', 'es.port'='9200', 'es.mapping.id' ='id') 注： 1. es.resource对应es中的index/type 2. 1.es.mapping.names为hive和es字段名映射关系。 2.如果hive表和es表字段名完全一致，可以省略此参数。 3.hive中字段名不区分大小写，元数据寸的全是小写；es中字段大小写敏感，如果es中字段名出现大写，需认真填写。 4.es中_id为自动生成，如若需要覆盖，需加参数'es.mapping.id'='id' 2 先导入es映射表相关jar包 add jar /opt/cloudera/parcels/CDH-6.3.2-1.cdh6.3.2.p0.1605554/lib/hive/auxlib/elasticsearch-hadoop-6.3.0.jar; add jar /data/jar/httpclient-4.5.5.jar; add jar /data/jar/org.apache.commons.httpclient.jar; 3 向映射表insert数据 二 es导入hive 1 建hive映射表 CREATE EXTERNAL TABLE hive_es.cty_test5( addTime string ) STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler' TBLPROPERTIES('es.resource' = 'cty_test/cty_test', 'es.nodes'='172.16.98.113,172.16.98.149,172.16.98.150,172.16.98.151,172.16.98.152', 'es.port'='9200', 'es.mapping.names'= 'addTime:addTime', 'es.mapping.date.rich'='false', 'es.index.auto.create'='false', ) 注意，hive表数据类型要和es一致，除了es的date要转成hive的string，同时要加参数’es.mapping.date.rich’=‘false’,否则查询会报错. 2 通过映射表向其他表insert 参考： https://www.cnblogs.com/koushr/p/9505435.html

set spark.master=yarn-cluster; #设置spark提交模式 set hive.execution.engine=spark; #设置计算引擎 set spark.yarn.queue=queue_name; #设置作业提交队列 set spark.app.name=job_name; #设置作业名称 set spark.executor.instances=20; #设置执行器个数 set spark.executor.cores=4; #设置执行器计算核个数 set spark.executor.memory=8g; #设置执行器内存 set mapred.reduce.tasks=600; #设置任务并行度 set spark.yarn.executor.memoryOverhead=2048; #设置每个executor的jvm堆外内存 set spark.memory.fraction=0.8; #设置内存比例(spark2.0+) set spark.serializer=org.apache.serializer.KyroSerializer; #设置对象序列化方式

调优具体细节 Hive建表设计层面 Hive 的建表设计层面调优，主要讲的怎么样合理的组织数据，方便后续的高效计算。比如建表的类型，文件存储格式，是否压缩等等。 利用分区表优化 关于 Hive 的表的类型有哪些？ 1、分区表 2、分桶表 分区表 是在某一个或者几个维度上对数据进行分类存储，一个分区对应一个目录。如果筛选条件里有分区字段，那么 Hive 只需要遍历对应分区目录下的文件即可，不需要遍历全局数据，使得处理的数据量大大减少，从而提高查询效率。 也就是说：当一个 Hive 表的查询大多数情况下，会根据某一个字段进行筛选时，那么非常适合创建为分区表，该字段即为分区字段。 select1: select .... where country = "china" select2: select .... where country = "china" select3: select .... where country = "china" select4: select .... where country = "china" ..... 分门别类：这个city字段的每个值，就单独形成为一个分区。其实每个分区就对应带HDFS的一个目录 在创建表时通过启用 partitioned by 实现，用来 partition 的维度并不是实际数据的某一列，具体分区的标志是由插入内容时给定的。当要查询某一分区的内容时可以采用 where 语句，形似 where tablename.partition_column = a 来实现。 1、创建含分区的表： CREATE TABLE page_view(viewTime INT, userid BIGINT, page_url STRING, referrer_url STRING, ip STRING COMMENT 'IP Address of the User') PARTITIONED BY(date STRING, country STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY '1' STORED AS TEXTFILE; 2、载入内容，并指定分区标志： ...

1.动态分区 SET hive.exec.dynamic.partition=true; SET hive.exec.dynamic.partition.mode=nonstrict; SET hive.exec.max.dynamic.partitions=100000; SET hive.exec.max.dynamic.partitions.pernode=100000; SET hive.exec.max.created.files=100000; 2.union all并发执行 --在使用union all的时候，系统资源足够的情况下，为了加快hive处理速度，可以设置如下参数实现并发执行 set mapred.job.priority=VERY_HIGH; set hive.exec.parallel=true; 3.设置map reduce个数 -- 设置map capacity set mapred.job.map.capacity=2000; set mapred.job.reduce.capacity=2000; -- 设置每个reduce的大小 set hive.exec.reducers.bytes.per.reducer=500000000; -- 直接设置个数 set mapred.reduce.tasks = 15; 4.文件合并 -- 设置文件合并 set abaci.is.dag.job=false; set hive.merge.mapredfiles=true; set mapred.combine.input.format.local.only=false; set hive.merge.smallfiles.avgsize=100000000; -- 在map only的情况下，如上的参数如果没有生效，可以设置如下 -- 在HQL的最外层增加distribute by rand() select * from XXX distribute by rand() 5.设置任务名称 -- 设置名称 set mapred.job.name=${my_job}; 6.设置引擎和指定队列 set hive.execution.engine=mr; set mapreduce.job.queuename=bigdata;

1.MongoDB拉出到hive # 导入mongodb的包到hadoop add jar /var/lib/hadoop-hdfs/bin/hive_mongoDB/mongo-hadoop-core-2.0.2.jar; add jar /var/lib/hadoop-hdfs/bin/hive_mongoDB/mongo-hadoop-hive-2.0.2.jar; add jar /var/lib/hadoop-hdfs/bin/hive_mongoDB/mongo-java-driver-3.12.8.jar; set mongo.input.split.create_input_splits=false; DROP TABLE IF EXISTS ods.ods_ex_trade_mdb_wmb_rocket_waimao_company_info_mapping; drop table ods.ods_zxk_annualBusiness_mapping; create external table ods.ods_zxk_annualBusiness_mapping( `_id` string, companyName string, Tel string )STORED BY 'com.mongodb.hadoop.hive.MongoStorageHandler' WITH SERDEPROPERTIES('mongo.columns.mapping'='{"_id":"_id","companyName":"companyName","Tel":"Tel"}') TBLPROPERTIES('mongo.uri'='mongodb://username:password@172.16.98.159:21000/annualReport.annualBusiness?authSource=admin');