兰州百度网站建设,新网站建设哪家好,网页制作教程零基础合集,珠海公司网站设计简介
Hive 默认是不允许数据更新操作的#xff0c;毕竟它不擅长#xff0c;即使在0.14版本后#xff0c;做一些额外的配置便可开启Hive数据更新操作。而在海量数据场景下做update、delete之类的行级数据操作时#xff0c;效率并不如意。
简单使用
修改HIVE_HOME/conf/hi…简介
Hive 默认是不允许数据更新操作的毕竟它不擅长即使在0.14版本后做一些额外的配置便可开启Hive数据更新操作。而在海量数据场景下做update、delete之类的行级数据操作时效率并不如意。
简单使用
修改HIVE_HOME/conf/hive-site.xml添加如下配置
propertynamehive.support.concurrency/namevaluetrue/value
/property
propertynamehive.exec.dynamic.partition.mode/namevaluenonstrict/value
/property
propertynamehive.txn.manager/namevalueorg.apache.hadoop.hive.ql.lockmgr.DbTxnManager/value
/property
propertynamehive.compactor.initiator.on/namevaluetrue/value
/property
propertynamehive.compactor.worker.threads/namevalue1/value
/property
建表
create table if not exists accountInfo(
id int,
name string,
age int
)
clustered by (id) into 4 buckets
stored as orc TBLPROPERTIES (transactionaltrue);建表须知 1 注意存储格式按ORC方式 2 进行数据分桶 3 添加表属性‘transactional’‘true’ 分发配置到其他hive节点。
测试如下
插入测试数据 更新数据此处配置了Hive on Tez 数据删除 hive作为数仓常用技术工具更多的是用于数据的存储分析而比较少涉及到数据更新。并且在OLAP场景下并不适合做原有数据更新更不用说行级别的细粒度操作。记得在一些状态更新场景下会有缓慢渐变维的运用可即使如此也要运用拉链表保存历史数据很少将原有数据直接覆盖你不知道被覆盖的数据蕴含着怎样的价值。
而在一些海量OLTP场景中也会运用Hbase去替代传统RDB架构若在运用中伴有大量的数据更新操作我想Hbase会是不错的选择。 Hive事务原理简介
Apache Hive 0.13 版本引入了事务特性能够在 Hive 表上实现 ACID 语义包括 INSERT/UPDATE/DELETE/MERGE 语句、增量数据抽取等。Hive 3.0 又对该特性进行了优化包括改进了底层的文件组织方式减少了对表结构的限制以及支持条件下推和向量化查询。Hive 事务表的介绍和使用方法可以参考 Hive Wiki 和 各类教程本文将重点讲述 Hive 事务表是如何在 HDFS 上存储的及其读写过程是怎样的。
文件结构
插入数据 1
2
3
4
5
6
7CREATE TABLE employee (id int, name string, salary int)
STORED AS ORC TBLPROPERTIES (transactional true);INSERT INTO employee VALUES
(1, Jerry, 5000),
(2, Tom, 8000),
(3, Kate, 6000);INSERT 语句会在一个事务中运行。它会创建名为 delta 的目录存放事务的信息和表的数据。 1
2
3/user/hive/warehouse/employee/delta_0000001_0000001_0000
/user/hive/warehouse/employee/delta_0000001_0000001_0000/_orc_acid_version
/user/hive/warehouse/employee/delta_0000001_0000001_0000/bucket_00000
目录名称的格式为 delta_minWID_maxWID_stmtID即 delta 前缀、写事务的 ID 范围、以及语句 ID。具体来说
所有 INSERT 语句都会创建 delta 目录。UPDATE 语句也会创建 delta 目录但会先创建一个 delete 目录即先删除、后插入。delete 目录的前缀是 delete_deltaHive 会为所有的事务生成一个全局唯一的 ID包括读操作和写操作。针对写事务INSERT、DELETE 等Hive 还会创建一个写事务 IDWrite ID该 ID 在表范围内唯一。写事务 ID 会编码到 delta 和 delete 目录的名称中语句 IDStatement ID则是当一个事务中有多条写入语句时使用的用作唯一标识。
再看文件内容_orc_acid_version 的内容是 2即当前 ACID 版本号是 2。它和版本 1 的主要区别是 UPDATE 语句采用了 split-update 特性即上文提到的先删除、后插入。这个特性能够使 ACID 表支持条件下推等功能具体可以查看 HIVE-14035。bucket_00000 文件则是写入的数据内容。由于这张表没有分区和分桶所以只有这一个文件。事务表都以 ORC 格式存储的我们可以使用 orc-tools 来查看文件的内容 1
2
3
4$ orc-tools data bucket_00000
{operation:0,originalTransaction:1,bucket:536870912,rowId:0,currentTransaction:1,row:{id:1,name:Jerry,salary:5000}}
{operation:0,originalTransaction:1,bucket:536870912,rowId:1,currentTransaction:1,row:{id:2,name:Tom,salary:8000}}
{operation:0,originalTransaction:1,bucket:536870912,rowId:2,currentTransaction:1,row:{id:3,name:Kate,salary:6000}}输出内容被格式化为了一行行的 JSON 字符串我们可以看到具体数据是在 row 这个键中的其它键则是 Hive 用来实现事务特性所使用的具体含义为
operation 0 表示插入1 表示更新2 表示删除。由于使用了 split-updateUPDATE 是不会出现的originalTransaction 是该条记录的原始写事务 ID。对于 INSERT 操作该值和 currentTransaction 是一致的。对于 DELETE则是该条记录第一次插入时的写事务 IDbucket 是一个 32 位整型由 BucketCodec 编码各个二进制位的含义为 1-3 位编码版本当前是 0014 位保留5-16 位分桶 ID由 0 开始。分桶 ID 是由 CLUSTERED BY 子句所指定的字段、以及分桶的数量决定的。该值和 bucket_N 中的 N 一致17-20 位保留21-32 位语句 ID举例来说整型 536936448 的二进制格式为 00100000000000010000000000000000即它是按版本 1 的格式编码的分桶 ID 为 1rowId 是一个自增的唯一 ID在写事务和分桶的组合中唯一currentTransaction 当前的写事务 IDrow 具体数据。对于 DELETE 语句则为 null。
我们可以注意到文件中的数据会按 (originalTransaction, bucket, rowId) 进行排序这点对后面的读取操作非常关键。
这些信息还可以通过 row__id 这个虚拟列进行查看 1SELECT row__id, id, name, salary FROM employee;输出结果为 1
2
3{writeid:1,bucketid:536870912,rowid:0} 1 Jerry 5000
{writeid:1,bucketid:536870912,rowid:1} 2 Tom 8000
{writeid:1,bucketid:536870912,rowid:2} 3 Kate 6000增量数据抽取 API V2
Hive 3.0 还改进了先前的 增量抽取 API通过这个 API用户或第三方工具Flume 等就可以利用 ACID 特性持续不断地向 Hive 表写入数据了。这一操作同样会生成 delta 目录但更新和删除操作不再支持。 1
2
3
4
5
6StreamingConnection connection HiveStreamingConnection.newBuilder().connect();
connection.beginTransaction();
connection.write(11,val11,Asia,China.getBytes());
connection.write(12,val12,Asia,India.getBytes());
connection.commitTransaction();
connection.close();更新数据 1UPDATE employee SET salary 7000 WHERE id 2;这条语句会先查询出所有符合条件的记录获取它们的 row__id 信息然后分别创建 delete 和 delta 目录 1
2
3/user/hive/warehouse/employee/delta_0000001_0000001_0000/bucket_00000
/user/hive/warehouse/employee/delete_delta_0000002_0000002_0000/bucket_00000
/user/hive/warehouse/employee/delta_0000002_0000002_0000/bucket_00000delete_delta_0000002_0000002_0000/bucket_00000 包含了删除的记录 1{operation:2,originalTransaction:1,bucket:536870912,rowId:1,currentTransaction:2,row:null}delta_0000002_0000002_0000/bucket_00000 包含更新后的数据 1{operation:0,originalTransaction:2,bucket:536870912,rowId:0,currentTransaction:2,row:{id:2,name:Tom,salary:7000}}DELETE 语句的工作方式类似同样是先查询后生成 delete 目录。
合并表
MERGE 语句和 MySQL 的 INSERT ON UPDATE 功能类似它可以将来源表的数据合并到目标表中 1
2
3
4
5
6
7
8
9CREATE TABLE employee_update (id int, name string, salary int);
INSERT INTO employee_update VALUES
(2, Tom, 7000),
(4, Mary, 9000);MERGE INTO employee AS a
USING employee_update AS b ON a.id b.id
WHEN MATCHED THEN UPDATE SET salary b.salary
WHEN NOT MATCHED THEN INSERT VALUES (b.id, b.name, b.salary);这条语句会更新 Tom 的薪资字段并插入一条 Mary 的新记录。多条 WHEN 子句会被视为不同的语句有各自的语句 IDStatement ID。INSERT 子句会创建 delta_0000002_0000002_0000 文件内容是 Mary 的数据UPDATE 语句则会创建 delete_delta_0000002_0000002_0001 和 delta_0000002_0000002_0001 两个文件删除并新增 Tom 的数据。 1
2
3
4/user/hive/warehouse/employee/delta_0000001_0000001_0000
/user/hive/warehouse/employee/delta_0000002_0000002_0000
/user/hive/warehouse/employee/delete_delta_0000002_0000002_0001
/user/hive/warehouse/employee/delta_0000002_0000002_0001压缩
随着写操作的积累表中的 delta 和 delete 文件会越来越多。事务表的读取过程中需要合并所有文件数量一多势必会影响效率。此外小文件对 HDFS 这样的文件系统也是不够友好的。因此Hive 引入了压缩Compaction的概念分为 Minor 和 Major 两类。
Minor Compaction 会将所有的 delta 文件压缩为一个文件delete 也压缩为一个。压缩后的结果文件名中会包含写事务 ID 范围同时省略掉语句 ID。压缩过程是在 Hive Metastore 中运行的会根据一定阈值自动触发。我们也可以使用如下语句人工触发 1ALTER TABLE employee COMPACT minor;以上文中的 MERGE 语句的结果举例在运行了一次 Minor Compaction 后文件目录结构将变为 1
2/user/hive/warehouse/employee/delete_delta_0000001_0000002
/user/hive/warehouse/employee/delta_0000001_0000002在 delta_0000001_0000002/bucket_00000 文件中数据会被排序和合并起来因此文件中将包含两行 Tom 的数据。Minor Compaction 不会删除任何数据。
Major Compaction 则会将所有文件合并为一个文件以 base_N 的形式命名其中 N 表示最新的写事务 ID。已删除的数据将在这个过程中被剔除。row__id 则按原样保留。 1/user/hive/warehouse/employee/base_0000002需要注意的是在 Minor 或 Major Compaction 执行之后原来的文件不会被立刻删除。这是因为删除的动作是在另一个名为 Cleaner 的线程中执行的。因此表中可能同时存在不同事务 ID 的文件组合这在读取过程中需要做特殊处理。
读取过程
我们可以看到 ACID 事务表中会包含三类文件分别是 base、delta、以及 delete。文件中的每一行数据都会以 row__id 作为标识并排序。从 ACID 事务表中读取数据就是对这些文件进行合并从而得到最新事务的结果。这一过程是在 OrcInputFormat 和 OrcRawRecordMerger 类中实现的本质上是一个合并排序的算法。
以下列文件为例产生这些文件的操作为插入三条记录进行一次 Major Compaction然后更新两条记录。1-0-0-1 是对 originalTransaction - bucketId - rowId - currentTransaction 的缩写。 1
2
3
4
5
6
7---------- ---------- ----------
| base_1 | | delete_2 | | delta_2 |
---------- ---------- ----------
| 1-0-0-1 | | 1-0-1-2 | | 2-0-0-2 |
| 1-0-1-1 | | 1-0-2-2 | | 2-0-1-2 |
| 1-0-2-1 | ---------- ----------
----------合并过程为
对所有数据行按照 (originalTransaction, bucketId, rowId) 正序排列(currentTransaction) 倒序排列即 1-0-0-11-0-1-21-0-1-1…2-0-1-2获取第一条记录如果当前记录的 row__id 和上条数据一样则跳过如果当前记录的操作类型为 DELETE也跳过 通过以上两条规则对于 1-0-1-2 和 1-0-1-1这条记录会被跳过如果没有跳过记录将被输出给下游重复以上过程。
合并过程是流式的即 Hive 会将所有文件打开预读第一条记录并将 row__id 信息存入到 ReaderKey 类型中。该类型实现了 Comparable 接口因此可以按照上述规则自定义排序 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30public class RecordIdentifier implements WritableComparableRecordIdentifier {private long writeId;private int bucketId;private long rowId;protected int compareToInternal(RecordIdentifier other) {if (other null) { return -1; }if (writeId ! other.writeId) { return writeId other.writeId ? -1 : 1; }if (bucketId ! other.bucketId) { return bucketId other.bucketId ? - 1 : 1; }if (rowId ! other.rowId) { return rowId other.rowId ? -1 : 1; }return 0;}
}public class ReaderKey extends RecordIdentifier {private long currentWriteId;private boolean isDeleteEvent false;public int compareTo(RecordIdentifier other) {int sup compareToInternal(other);if (sup 0) {if (other.getClass() ReaderKey.class) {ReaderKey oth (ReaderKey) other;if (currentWriteId ! oth.currentWriteId) { return currentWriteId oth.currentWriteId ? 1 : -1; }if (isDeleteEvent ! oth.isDeleteEvent) { return isDeleteEvent ? -1 : 1; }} else {return -1;}}return sup;}
}然后ReaderKey 会和文件句柄一起存入到 TreeMap 结构中。根据该结构的特性我们每次获取第一个元素时就能得到排序后的结果并读取数据了。 1
2
3
4
5
6public class OrcRawRecordMerger {private TreeMapReaderKey, ReaderPair readers new TreeMap();public boolean next(RecordIdentifier recordIdentifier, OrcStruct prev) {Map.EntryReaderKey, ReaderPair entry readers.pollFirstEntry();}
}选择文件
上文中提到事务表目录中会同时存在多个事务的快照文件因此 Hive 首先要选择出反映了最新事务结果的文件集合然后再进行合并。举例来说下列文件是一系列操作后的结果两次插入一次 Minor Compaction一次 Major Compaction一次删除。 1
2
3
4
5delta_0000001_0000001_0000
delta_0000002_0000002_0000
delta_0000001_0000002
base_0000002
delete_delta_0000003_0000003_0000过滤过程为
从 Hive Metastore 中获取所有成功提交的写事务 ID 列表从文件名中解析出文件类型、写事务 ID 范围、以及语句 ID选取写事务 ID 最大且合法的那个 base 目录如果存在的话对 delta 和 delete 文件进行排序 minWID 较小的优先如果 minWID 相等则 maxWID 较大的优先如果都相等则按 stmtID 排序没有 stmtID 的会排在前面将 base 文件中的写事务 ID 作为当前 ID循环过滤所有 delta 文件 如果 maxWID 大于当前 ID则保留这个文件并以此更新当前 ID如果 ID 范围相同也会保留这个文件重复上述步骤。
过滤过程中还会处理一些特别的情况如没有 base 文件有多条语句包含原始文件即不含 row__id 信息的文件一般是通过 LOAD DATA 导入的以及 ACID 版本 1 格式的文件等。具体可以参考 AcidUtils#getAcidState 方法。
并行执行
在 Map-Reduce 模式下运行 Hive 时多个 Mapper 是并行执行的这就需要将 delta 文件按一定的规则组织好。简单来说base 和 delta 文件会被分配到不同的分片Split中但所有分片都需要能够读取所有的 delete 文件从而根据它们忽略掉已删除的记录。 向量化查询
当 向量化查询 特性开启时Hive 会尝试将所有的 delete 文件读入内存并维护一个特定的数据结构能够快速地对数据进行过滤。如果内存放不下则会像上文提到的过程一样逐步读取 delete 文件使用合并排序的算法进行过滤。 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20public class VectorizedOrcAcidRowBatchReader {private final DeleteEventRegistry deleteEventRegistry;protected static interface DeleteEventRegistry {public void findDeletedRecords(ColumnVector[] cols, int size, BitSet selectedBitSet);}static class ColumnizedDeleteEventRegistry implements DeleteEventRegistry {}static class SortMergedDeleteEventRegistry implements DeleteEventRegistry {}public boolean next(NullWritable key, VectorizedRowBatch value) {BitSet selectedBitSet new BitSet(vectorizedRowBatchBase.size);this.deleteEventRegistry.findDeletedRecords(innerRecordIdColumnVector,vectorizedRowBatchBase.size, selectedBitSet);for (int setBitIndex selectedBitSet.nextSetBit(0), selectedItr 0;setBitIndex 0;setBitIndex selectedBitSet.nextSetBit(setBitIndex1), selectedItr) {value.selected[selectedItr] setBitIndex;}}
}事务管理
为了实现 ACID 事务机制Hive 还引入了新的事务管理器 DbTxnManager它能够在查询计划中分辨出 ACID 事务表联系 Hive Metastore 打开新的事务完成后提交事务。它也同时实现了过去的读写锁机制用来支持非事务表的情形。 Hive Metastore 负责分配新的事务 ID。这一过程是在一个数据库事务中完成的从而避免多个 Metastore 实例冲突的情况。 1
2
3
4
5
6
7
8
9
10
11
12
13abstract class TxnHandler {private ListLong openTxns(Connection dbConn, Statement stmt, OpenTxnRequest rqst) {String s sqlGenerator.addForUpdateClause(select ntxn_next from NEXT_TXN_ID);s update NEXT_TXN_ID set ntxn_next (first numTxns);for (long i first; i first numTxns; i) {txnIds.add(i);rows.add(i , quoteChar(TXN_OPEN) , now , now , quoteString(rqst.getUser()) , quoteString(rqst.getHostname()) , txnType.getValue());}ListString queries sqlGenerator.createInsertValuesStmt(TXNS (txn_id, txn_state, txn_started, txn_last_heartbeat, txn_user, txn_host, txn_type), rows);}
} PS: 向量化查询 向量化查询执行是Hive特性可以大大减少典型查询操作如扫描过滤器聚合和连接的CPU使用率。一个标准的查询执行系统一次处理一行。这涉及在执行的内部循环中长的代码路径和重要的元数据解释。目前Hive也严重依赖于惰性的反序列化数据列通过一层对象检查器来识别列类型反序列化数据并在内部循环中确定合适的表达式例程。这些虚拟方法调用层进一步减慢了处理速度。向量化的查询执行通过一次处理1024行的数据块来简化操作。在块内每一列都被存储为一个向量一个基本数据类型的数组。算术和比较等简单操作是通过在紧密循环中快速迭代向量来完成的在循环内没有或很少有函数调用或条件分支。这些循环以简化的方式进行编译使用相对较少的指令并通过有效地使用处理器流水线和高速缓存存储器以较少的时钟周期完成每条指令 文章转载自： http://www.morning.ksqzd.cn.gov.cn.ksqzd.cn http://www.morning.ryxdr.cn.gov.cn.ryxdr.cn http://www.morning.chbcj.cn.gov.cn.chbcj.cn http://www.morning.hpdpp.cn.gov.cn.hpdpp.cn http://www.morning.kyctc.cn.gov.cn.kyctc.cn http://www.morning.nqbs.cn.gov.cn.nqbs.cn http://www.morning.bnfsw.cn.gov.cn.bnfsw.cn http://www.morning.kxbdm.cn.gov.cn.kxbdm.cn http://www.morning.lgmgn.cn.gov.cn.lgmgn.cn http://www.morning.ldynr.cn.gov.cn.ldynr.cn http://www.morning.jzccn.cn.gov.cn.jzccn.cn http://www.morning.gtbjf.cn.gov.cn.gtbjf.cn http://www.morning.qmzhy.cn.gov.cn.qmzhy.cn http://www.morning.rkfgx.cn.gov.cn.rkfgx.cn http://www.morning.rrgm.cn.gov.cn.rrgm.cn http://www.morning.bykqg.cn.gov.cn.bykqg.cn http://www.morning.sqmbb.cn.gov.cn.sqmbb.cn http://www.morning.yqqxj26.cn.gov.cn.yqqxj26.cn http://www.morning.jnvivi.com.gov.cn.jnvivi.com http://www.morning.hmxrs.cn.gov.cn.hmxrs.cn http://www.morning.dndk.cn.gov.cn.dndk.cn http://www.morning.qnbgk.cn.gov.cn.qnbgk.cn http://www.morning.fbbmg.cn.gov.cn.fbbmg.cn http://www.morning.nwynx.cn.gov.cn.nwynx.cn http://www.morning.lywys.cn.gov.cn.lywys.cn http://www.morning.qpqb.cn.gov.cn.qpqb.cn http://www.morning.bkylg.cn.gov.cn.bkylg.cn http://www.morning.hbfqm.cn.gov.cn.hbfqm.cn http://www.morning.mcjxq.cn.gov.cn.mcjxq.cn http://www.morning.fjntg.cn.gov.cn.fjntg.cn http://www.morning.cbqqz.cn.gov.cn.cbqqz.cn http://www.morning.ywpwq.cn.gov.cn.ywpwq.cn http://www.morning.skrww.cn.gov.cn.skrww.cn http://www.morning.bgqqr.cn.gov.cn.bgqqr.cn http://www.morning.rcwzf.cn.gov.cn.rcwzf.cn http://www.morning.ylmxs.cn.gov.cn.ylmxs.cn http://www.morning.qdxwf.cn.gov.cn.qdxwf.cn http://www.morning.dwtdn.cn.gov.cn.dwtdn.cn http://www.morning.gycyt.cn.gov.cn.gycyt.cn http://www.morning.bnxnq.cn.gov.cn.bnxnq.cn http://www.morning.zstry.cn.gov.cn.zstry.cn http://www.morning.uycvv.cn.gov.cn.uycvv.cn http://www.morning.ptwzy.cn.gov.cn.ptwzy.cn http://www.morning.prmbn.cn.gov.cn.prmbn.cn http://www.morning.ztqj.cn.gov.cn.ztqj.cn http://www.morning.bsghk.cn.gov.cn.bsghk.cn http://www.morning.snyqb.cn.gov.cn.snyqb.cn http://www.morning.nqfxq.cn.gov.cn.nqfxq.cn http://www.morning.kybjr.cn.gov.cn.kybjr.cn http://www.morning.zkqsc.cn.gov.cn.zkqsc.cn http://www.morning.fhbhr.cn.gov.cn.fhbhr.cn http://www.morning.nkcfh.cn.gov.cn.nkcfh.cn http://www.morning.yrpd.cn.gov.cn.yrpd.cn http://www.morning.xbyyd.cn.gov.cn.xbyyd.cn http://www.morning.rtsdz.cn.gov.cn.rtsdz.cn http://www.morning.jxrpn.cn.gov.cn.jxrpn.cn http://www.morning.kndt.cn.gov.cn.kndt.cn http://www.morning.qhrdx.cn.gov.cn.qhrdx.cn http://www.morning.cjqqj.cn.gov.cn.cjqqj.cn http://www.morning.ftntr.cn.gov.cn.ftntr.cn http://www.morning.wnpps.cn.gov.cn.wnpps.cn http://www.morning.plznfnh.cn.gov.cn.plznfnh.cn http://www.morning.mszwg.cn.gov.cn.mszwg.cn http://www.morning.nkjnr.cn.gov.cn.nkjnr.cn http://www.morning.kfmlf.cn.gov.cn.kfmlf.cn http://www.morning.caswellintl.com.gov.cn.caswellintl.com http://www.morning.gtxrw.cn.gov.cn.gtxrw.cn http://www.morning.rbjth.cn.gov.cn.rbjth.cn http://www.morning.beijingzy.com.cn.gov.cn.beijingzy.com.cn http://www.morning.xxwhz.cn.gov.cn.xxwhz.cn http://www.morning.zyytn.cn.gov.cn.zyytn.cn http://www.morning.fbbmg.cn.gov.cn.fbbmg.cn http://www.morning.kqblk.cn.gov.cn.kqblk.cn http://www.morning.mpszk.cn.gov.cn.mpszk.cn http://www.morning.yrblz.cn.gov.cn.yrblz.cn http://www.morning.dhpjq.cn.gov.cn.dhpjq.cn http://www.morning.gqwpl.cn.gov.cn.gqwpl.cn http://www.morning.sxwfx.cn.gov.cn.sxwfx.cn http://www.morning.yaqi6.com.gov.cn.yaqi6.com http://www.morning.rdng.cn.gov.cn.rdng.cn
