Hive的增量更新策略：四步

1.INCREMENTAL UPDATES

       Hadoop和Hive正在快速发展，以超越以前集成和数据访问的限制。在近期的开发路线图上，我们希望Hive支持完整的CRUD操作。当我们等待这些更改时，仍然需要使用当前选项——OVERWRITE或APPEND来进行Hive表集成。
       OVERWRITE选项要求将完整的记录集从源端移动到hadoop。虽然这种方法可能适用于较小数据集，但一般是禁止的。
       APPEND选项会对数据进行限制，只包含new或update的记录。由于Hive中还没有真正的插入和更新，我们需要考虑一个防止重复记录的过程，因为update被附加到累计的记录集中。
       本文中，我们将研究从RDBMS源向现有Hive表定义添加update和insert的4步策略。

IMPORTANT：本文中，我们假设每个源表都有唯一的单键或多键，并且为每个记录维护一个“modified_date”字段——要么定义为原始源表的一部分，要么在抽取时添加。

2.HIVE TABLE DEFINITION OPTIONS:EXTERNAL, LOCAL AND VIEW

       外部表是Hive表定义和HDFS管理的文件夹和文件的组合。表定义独立于数据存在，因此，如果删除表，HDFS文件夹和文件将保持原始状态。
       本地表是直接绑定到源数据的Hive表。数据在物理上绑定到表定义，如果表被删除，数据也将被删除。
       与传统的RDBMS一样，视图是存储的SQL查询，支持与Hive表相同的READ交互，但是它们不存储自己的任何数据。相反，数据从存储的对Hive表的SQL查询中获取。
       以下流程概述了一个工作流，该工作流利用了上述所有的四个步骤：

image.png

• Ingest（抽取）。从完整的表（base table）抽取更改记录（incremental_table）。
• Reconcile（一致）。创建一个Base+Change records（reconcile_view）的单一视图，反应最新的记录集。
• Compact（结合）。从reconcile视图创建报表（reporting_table）。
• Purge（净化）。使用reporting table替换base table，并在下一个数据抽取周期前删除以前处理过的更改记录。

       将作为增量更新工作流的表和视图有：

• base_table：一个HIVE本地表，最初保存来自源系统的所有记录。在初始处理周期后，它将维护来自源的最新同步记录集的副本。在每个处理周期后，reporting_table将覆盖它。（Step 4：Purge）。
• incremental_table：一个Hive外部表，保存来自源系统的增量更改记录（插入和更新）。在每个处理周期结束时，清除其中内存（Step 4：Purge）
• reconcile_view：一个Hive视图，组合并产生base_table和incremental_table内容，只显示最新记录。用于填充reporting_table（Step 3：Compact）
• reporting_table：一个Hive本地表，其中包含目标的最新记录。用来对base_table进行覆盖。

STEP 1:INGEST

       根据RDBMS源系统是否提供直接访问，可以选择文件处理方法（直接访问不可用时）或RDBMS处理（直接访问可用时）。
       无论选择哪种，都需要以下工作流：
       1.一次性，初始加载所有数据到HIVE。
       2.不断加载变化数据到HIVE。
       下面将会讨论文件处理和数据库直接处理（Sqoop）。

FILE PROCESSING

       本文中，我们假设文件夹中的文件具有分隔格式，且将由关系系统产生（即记录具有唯一的键或标识符）。
       文件通过以下两种方式移动到HDFS：

• WebHDFS：主要用于与应用集成，Web URL提供一个上载端点到指定HDFS文件夹中。
• NFS：显示为标准网络驱动器，并允许终端用户使用标准复制粘贴操作将文件从标准文件系统移动到HDFS。
         一旦将初始记录集移动到HDFS中，后续就可以只移动新插入的更新的记录。

RDBMS PROCESSING

       SQOOP是基于JDBC的集成传统数据库的工具。SQOOP允许将数据移动到HDFS，或直接移动到Hive表中。
       可以使用-table参数将整个源表移动到HDFS或Hive中。

sqoop import --connect jdbc:teradata://{host name or ip address}/Database=retail
--connection-manager org.apache.sqoop.teradata.TeradataConnManager
--username dbc
--password dbc
--table SOURCE_TBL
--target-dir /user/hive/incremental_table -m 1

       在初始导入之后，后续导入可以使用“check-column”、“incremental”和“last-value”参数利用SQOOP对增量导入的支持。

sqoop import --connect jdbc:teradata://{host name or ip address}/Database=retail
--connection-manager org.apache.sqoop.teradata.TeradataConnManager
--username dbc
--password dbc
--table SOURCE_TBL
--target-dir /user/hive/incremental_table
-m 1 --check-column modified_date
--incremetal lastmodified --last-value {last_import_date}

       另外，可以利用“query”参数，让SQL select语句将导入限制为只导入新记录或更改记录。

sqoop import --connect jdbc:teradata://{host name or ip address}/Database=retail
--connection-manager org.apache.sqoop.teradata.TeradataConnManager
--username dbc
--password dbc
--target-dir /user/hive/incremental_table
-m 1 --query 'select * from SOURCE_TBL where modified_date > {last_import_date} AND $CONDITIONS'

note：对于初始加载，将“base_table”替换为“incremental_table”。对所有后续加载，使用“incremental_table”。

STEP 2:RECONCILE

       为了支持HIVE中的当前记录和新的更改记录之间的持续协调，应该定义两个表:base_table和incremental_table。

BASE_TABLE

       下面的示例显示了Hive表“base_table”的DDL，它将包含位于“/user/hive/base_table”目录下的HDFS中的任何带分隔符的文件。这个表将包含来自源系统的初始、完整的记录加载。在第一次处理程序运行后，它将存放来自源系统的最新记录:

create table base_table(
id string,
field1 string,
field2 string,
modified_date string
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LOCATION '/user/hive/base_table';

INCREMENTAL_TABLE

create external table base_table(
id string,
field1 string,
field2 string,
modified_date string
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LOCATION '/user/hive/incremental_table';

RECONCILE_VIEW

       这个视图组合了来自基表(base_table)和Change表(incremental_table)的记录集，并且仅生成每个惟一“id”的最近的记录。
       定义如下：

CREATE VIEW reconcile_view AS
SELECT t1.* FROM 
(SELECT * FROM base_table UNION ALL SELECT * FROM incremental_table) t1
JOIN
(SELECT id, max(modified_date) max_modified FROM
SELECT * FROM base_table
UNION ALL
SELECT * FROM incremental_table) t2
GROUP BY id) s
ON t1.id = s.id AND t1.modified_date = s.max_modified;

STEP 3:COMPACT

       reconcile_view现在包含最新的记录集，并与来自RDBMS源系统的更改同步。对于BI报告和分析工具，reporting_table可以从reconcile_view生成。在创建此表之前，应该像下面示例一样删除表的任何以前的实例。

REPORTING_TABLE

DROP TABLE reporting_table;
CREATE TABLE reporting_table AS SELECT * FROM reconcile_view;

       将已协调视图(reconcile_view)移动到报表表(reporting_table)可以减少报表查询所需的处理量。
       此外，报表表中存储的数据也将是静态的，直到下一个处理周期才会更改。这提供了处理周期之间报告的一致性。相反，协调视图(reconcile_view)是动态的，只要在更改表(incremental_table)文件夹/user/hive/incremental_table中添加或删除新文件(保存更改记录)，它就会更改。

STEP 4:PURGE

       要准备来自源的下一系列增量记录，只需用最新的记录(reporting_table)替换基本表(base_table)。此外，通过删除位于外部表位置(' /user/hive/incremental_table ')的文件，删除以前导入的更改记录内容(incremental_table)。

DROP TABLE base_table;
CREATE TABLE base_table AS
SELECT * FROM reporting_table;

hadoop fs -rm -r /user/hive/incremental_table/*

FINAL THOUGHT

       虽然有几种可能的方法来支持增量数据馈送到Hive，这个例子有几个关键的优势:

• 通过维护只用于更新的外部表，可以通过简单地向该文件夹添加或删除文件来刷新表内容。
• 处理周期中的四个步骤(摄取、协调、压缩和清除)可以在一个OOZIE工作流中协调。OOZIE工作流可以是与数据更新SLA(即每日、每周、每月等)相对应的计划事件。
• 除了支持插入和更新同步之外，还可以通过向导入源添加DELETE_FLAG或DELETE_DATE字段来同步删除。然后，在Hive报表表中使用该字段作为过滤器来隐藏已删除的记录。例如,

CREATE VIEW reconcile_view AS
SELECT t1.* FROM
(SELECT * FROM base_table
UNION
SELECT * FROM incremental_table) t1
JOIN
(SELECT id, max(modified_date) max_modified FROM
(SELECT * FROM base_table
UNION
SELECT * FROM incremental_table)
GROUP BY id) s
ON t1.id = s.id AND t1.modified_date = s.max_modified
AND t1.delete_date IS NULL;