维护计划的查询

Intro

定期执行语句可能对

• 从外部系统获取信息

• 定期更新列统计信息

• 重建物化视图

Overview

• 元存储在元存储数据库中维护计划的查询

• Hiveserver 定期轮询元存储库以执行 sched 的查询

• 在执行期间，有关正在进行/完成的执行的信息保存在 metastore 中

Hive 在语言本身中内置了计划查询界面，以便于访问：

创建计划的查询语法

创建 sched 查询 <scheduled_query_name>
<scheduleSpecification>
[<executedAsSpec> ]
[<enableSpecification>]
<definedAsSpec>

更改 sched 的查询语法

更改计划的查询 <scheduled_query_name> (<scheduleSpec> | <executedAsSpec> | <enableSpecification> | <definedAsSpec> | <executeSpec>);

Drop syntax

计划的查询 <scheduled_query_name>;

scheduleSpecification syntax

时间表可以使用 CRON 表达式指定，或者在常见情况下可以使用更简单的形式；在任何情况下，计划都以 Quartz cron 表达式存储。

基于 CRON 的日程表语法

CRON <quartz_schedule_expression>

其中 quartz_schedule_expression 以 Quartz 格式引用了进度表

https://www.freeformatter.com/cron-expression-generator-quartz.html

例如，CRON '0 */10 * * * ? *'表达式将每 10 分钟触发一次。

每个基于计划的语法

为了提供一种更具可读性的声明时间表，可以使用每个。

每[(偏移量| AT) <timeOrDate>]

该格式使以更易读的方式声明时间表成为可能：

每 2 分钟
每小时(0:07:30)
每天'11：35：30'

ExecutedAs syntax

执行为 <user_name>

默认情况下，调度查询以声明用户的身份执行；但是具有 Management 员权限的人可能可以更改执行用户。

enableSpecification syntax

(启用[D] |禁用[D])

可用于启用/禁用时间表。

定义的 AS 语法

[已定义] AS

“查询”是要计划执行的单个语句表达式。

executeSpec syntax

EXECUTE

将计划的下一个执行时间更改为现在。在调试/开发过程中可能很有用。

System tables/views

可以使用* information_schema 或 sysdb *获取有关计划的查询/执行的信息-推荐的方法是使用 information_schema。 * sysdb 是用于构建 information_schema *级别视图的表-用于调试。

information_schema.scheduled_queries

假设我们有一个 sched 的查询，定义为：

**创建计划的查询 sc1 cron'0 */10 * * ？ '作为选择 1;

让我们在 information_schema.scheduled_queries 表中通过使用

从 information_schema.scheduled_queries 中选择；*

我将转置结果集以描述每一列

information_schema.scheduled_executions

该视图可用于获取有关最近计划的查询执行的信息。

从 information_schema.scheduled_executions 中选择;*

此视图中的一条记录具有以下信息：

Execution states

保留 INITED 状态，直到来自执行程序的第一次更新进入。
| EXECUTING |执行状态中的查询正在由执行者处理；在此阶段，执行程序按以下时间间隔报告查询进度：hive.scheduled.queries.executor.progress.report.interval ** |
| FAILED |设置了此状态时，由于错误代码(或异常)而停止了查询执行，并且还填充了 error_message 。
| 完成 |查询已成功完成，没有问题|
| TIMED_OUT |如果执行的执行时间超过 metastore.scheduled.queries.execution.timeout ，则该执行被视为超时。
计划的查询维护任务将检查是否有超时执行。

Configuration

Hive Metastore 相关配置

• metastore.scheduled.queries.enabled(默认值：true) **** 控制计划查询的 metastore 端支持；强制所有与 HMS 计划的查询相关的端点返回错误

• metastore.scheduled.queries.execution.timeout(默认值：2 分钟)
  如果在至少此时间量内未更新计划执行，请执行以下操作：它的状态将由清洁器任务更改为 TIMED_OUT

• ** metastore.scheduled.queries.execution.maint.task.frequency(默认值：1 分钟)
  计划查询维护任务的时间间隔。删除超过最高年龄的处决；并在满足条件的情况下将执行标记为 TIMED_OUT **

• metastore.scheduled.queries.execution.max.age(默认值：30 天)
  计划的查询执行条目在删除之前的最长期限。

HiveServer2 相关配置

• 启用 Hive 计划调度查询执行程序(默认：true)
  控制 HS2 是否将运行计划的查询执行程序。

• hive.scheduled.queries.namespace(默认值：“ hive”) **** 设置要使用的计划查询名称空间。在此命名空间中创建了新的计划查询。并且执行也绑定到名称空间

• hive.scheduled.queries.executor.idle.sleep.time(默认值：1 分钟)
  在查询计划的查询之间的睡眠时间。

• hive.scheduled.queries.executor.progress.report.interval(默认值：1 分钟) **** 在 sched 查询进行中；定期进行后台更新以报告查询的实际状态。

• hive.scheduled.queries.create.as.enabled(默认值：true) **** 此选项设置新创建的 sched 查询的默认行为。

• hive.security.authorization.scheduled.queries.supported(默认值：false) **** 如果配置的授权者能够处理与计划的查询相关的调用，请启用此功能。

Examples

示例 1 –使用计划的基本示例

create table t (a integer);

-- create a scheduled query; every 10 minute insert a new row
create scheduled query sc1 cron '0 */10 * * * ? *' as insert into t values (1);
-- depending on hive.scheduled.queries.create.as.enabled the query might get create in disabled mode
-- it can be enabled using:
alter scheduled query sc1 enabled;

-- inspect scheduled queries using the information_schema
select * from information_schema.scheduled_queries s where schedule_name='sc1';
+-----------------------+------------------+------------+----------------------+-------------------+---------+-----------+----------------------+
| s.scheduled_query_id  | s.schedule_name  | s.enabled  | s.cluster_namespace  |    s.schedule     | s.user  |  s.query  |   s.next_execution   |
+-----------------------+------------------+------------+----------------------+-------------------+---------+-----------+----------------------+
| 1                     | sc1              | true       | hive                 | 0 */10 * * * ? *  | dev     | select 1  | 2020-02-03 15:10:00  |
+-----------------------+------------------+------------+----------------------+-------------------+---------+-----------+----------------------+

-- wait 10 minutes or execute by issuing:
alter scheduled query sc1 execute;

select * from information_schema.scheduled_executions s where schedule_name='sc1' order by scheduled_execution_id desc limit 1;
+---------------------------+------------------+----------------------------------------------------+-----------+----------------------+----------------------+------------+------------------+---------------------+
| s.scheduled_execution_id  | s.schedule_name  |                s.executor_query_id                 |  s.state  |     s.start_time     |      s.end_time      | s.elapsed  | s.error_message  | s.last_update_time  |
+---------------------------+------------------+----------------------------------------------------+-----------+----------------------+----------------------+------------+------------------+---------------------+
| 496                       | sc1              | dev_20200203152025_bdf3deac-0ca6-407f-b122-c637e50f99c8 | FINISHED  | 2020-02-03 15:20:23  | 2020-02-03 15:20:31  | 8          | NULL             | NULL                |
+---------------------------+------------------+----------------------------------------------------+-----------+----------------------+----------------------+------------+------------------+---------------------+

示例 2 –定期分析外部表

假设您有一个外部表-它的内容正在缓慢变化...这最终将导致 Hive 在计划时间内使用过时的统计信息

-- create external table
create external table t (a integer);

-- see where the table lives:
desc formatted t;
[...]
| Location:                     | file:/data/hive/warehouse/t                       | NULL                                               |
[...]

-- in a terminal; load some data into the table directory:
seq 1 10 > /data/hive/warehouse/t/f1

-- back in hive you will see that 
select count(1) from t;
10
-- meanwhile basic stats show that the table has "0" rows 
desc formatted t;
[...]
|                               | numRows                                            | 0                                                  |
[...]

create scheduled query t_analyze cron '0 */1 * * * ? *' as analyze table t compute statistics for columns;

-- wait some time or execute by issuing:
alter scheduled query t_analyze execute;

select * from information_schema.scheduled_executions s where schedule_name='ex_analyze' order by scheduled_execution_id desc limit 3;
+---------------------------+------------------+----------------------------------------------------+------------+----------------------+----------------------+------------+------------------+----------------------+
| s.scheduled_execution_id  | s.schedule_name  |                s.executor_query_id                 |  s.state   |     s.start_time     |      s.end_time      | s.elapsed  | s.error_message  |  s.last_update_time  |
+---------------------------+------------------+----------------------------------------------------+------------+----------------------+----------------------+------------+------------------+----------------------+
| 498                       | t_analyze       | dev_20200203152640_a59bc198-3ed3-4ef2-8f63-573607c9914e | FINISHED   | 2020-02-03 15:26:38  | 2020-02-03 15:28:01  | 83         | NULL             | NULL                 |
+---------------------------+------------------+----------------------------------------------------+------------+----------------------+----------------------+------------+------------------+----------------------+

-- and the numrows have been updated
desc formatted t;
[...]
|                               | numRows                                            | 10                                                 |
[...]

-- we don't want this running every minute anymore...
alter scheduled query t_analyze disable;

示例 3 –重建实例化视图

-- some settings...they might be there already
set hive.support.concurrency=true;
set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
set hive.strict.checks.cartesian.product=false;
set hive.stats.fetch.column.stats=true;
set hive.materializedview.rewriting=true;

-- create some tables
CREATE TABLE emps (
  empid INT,
  deptno INT,
  name VARCHAR(256),
  salary FLOAT,
  hire_date TIMESTAMP)
STORED AS ORC
TBLPROPERTIES ('transactional'='true');
 
CREATE TABLE depts (
  deptno INT,
  deptname VARCHAR(256),
  locationid INT)
STORED AS ORC
TBLPROPERTIES ('transactional'='true');

-- load data
insert into emps values (100, 10, 'Bill', 10000, 1000), (200, 20, 'Eric', 8000, 500),
  (150, 10, 'Sebastian', 7000, null), (110, 10, 'Theodore', 10000, 250), (120, 10, 'Bill', 10000, 250),
  (1330, 10, 'Bill', 10000, '2020-01-02');
insert into depts values (10, 'Sales', 10), (30, 'Marketing', null), (20, 'HR', 20);

insert into emps values (1330, 10, 'Bill', 10000, '2020-01-02');

-- create mv
CREATE MATERIALIZED VIEW mv1 AS
  SELECT empid, deptname, hire_date FROM emps
    JOIN depts ON (emps.deptno = depts.deptno)
    WHERE hire_date >= '2016-01-01 00:00:00';

EXPLAIN
SELECT empid, deptname FROM emps
JOIN depts ON (emps.deptno = depts.deptno)
WHERE hire_date >= '2018-01-01';

-- create a schedule to rebuild mv
create scheduled query mv_rebuild cron '0 */10 * * * ? *' defined as 
  alter materialized view mv1 rebuild;

-- from this expalin it will be seen that the mv1 is being used
EXPLAIN
SELECT empid, deptname FROM emps
JOIN depts ON (emps.deptno = depts.deptno)
WHERE hire_date >= '2018-01-01';

-- insert a new record
insert into emps values (1330, 10, 'Bill', 10000, '2020-01-02');

-- the source tables are scanned
EXPLAIN
SELECT empid, deptname FROM emps
JOIN depts ON (emps.deptno = depts.deptno)
WHERE hire_date >= '2018-01-01';

-- wait 10 minutes or execute
alter scheduled query mv_rebuild execute;

-- run it again...the view should be rebuilt
EXPLAIN
SELECT empid, deptname FROM emps
JOIN depts ON (emps.deptno = depts.deptno)
WHERE hire_date >= '2018-01-01';

示例 4 –摄取

drop table if exists t;
drop table if exists s;

-- suppose that this table is an external table or something
-- which supports the pushdown of filter condition on the id column
create table s(id integer, cnt integer);

-- create an internal table and an offset table
create table t(id integer, cnt integer);
create table t_offset(offset integer);
insert into t_offset values(0);

-- pretend that data is added to s
insert into s values(1,1);

-- run an ingestion...
from (select id==offset as first,* from s
join t_offset on id>=offset) s1
insert into t select id,cnt where first = false
insert overwrite table t_offset select max(s1.id);

-- configure to run ingestion every 10 minutes
create scheduled query ingest every 10 minutes defined as
from (select id==offset as first,* from s
join t_offset on id>=offset) s1
insert into t select id,cnt where first = false
insert overwrite table t_offset select max(s1.id);

-- add some new values
insert into s values(2,2),(3,3);

-- pretend that a timeout have happened
alter scheduled query ingest execute;