Scheduled Queries

Scheduled Queries (예약 쿼리)

Scheduled Queries는 SQL 문을 주기적으로 자동 실행하는 Hive 4.0 이상의 기능이에요. 외부 시스템에서 정보를 가져오거나, 컬럼 통계를 주기적으로 갱신하거나, materialized view를 재구축하는 데 유용해요. metastore가 예약 정보를 저장하고 HiveServer가 주기적으로 실행할 예약을 폴링하죠.

출처: 문서

본문

소개 (Introduction)

문을 주기적으로 실행하는 것은 다음에 유용할 수 있어요.

• 외부 시스템에서 정보 가져오기
• 컬럼 통계 주기적으로 갱신
• materialized view 재구축

개요 (Overview)

• metastore는 예약 쿼리(scheduled queries)를 metastore 데이터베이스에 유지해요.
• HiveServer(s)는 주기적으로 metastore를 폴링해 실행할 예약 쿼리가 있는지 확인해요.
• 실행 중에는 진행 중/완료된 실행에 대한 정보가 metastore에 저장돼요.

Scheduled queries는 Hive 4.0에서 추가됐어요 (HIVE-21884).

Hive는 언어 자체에 예약 쿼리 인터페이스를 내장해 쉽게 접근할 수 있게 해요:

예약 쿼리 유지 관리 (Maintaining scheduled queries)

Create Scheduled query 구문

CREATE SCHEDULED QUERY <scheduled_query_name> `` [``] [``] ``

Alter Scheduled query 구문

ALTER SCHEDULED QUERY <scheduled_query_name> ( ``| ``| ``| ``| ``);

Drop 구문

DROP SCHEDULED QUERY <scheduled_query_name>;

scheduleSpecification 구문

일정은 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분마다 실행돼요.

EVERY 기반 일정 구문

일정을 더 읽기 쉽게 선언하기 위해 EVERY를 사용할 수 있어요.

EVERY [] (SECOND|MINUTE|HOUR) [(OFFSET BY|AT) ]

이 형식은 일정을 더 읽기 쉬운 방식으로 선언할 수 있게 해줘요:

EVERY 2 MINUTES EVERY HOUR AT '0:07:30' EVERY DAY AT '11:35:30'

ExecutedAs 구문

EXECUTED AS <user_name>

Scheduled queries는 기본적으로 선언한 사용자로 실행돼요. 하지만 관리 권한이 있는 사람은 실행 사용자를 변경할 수 있어요.

enableSpecification 구문

(ENABLE[D] | DISABLE[D])

일정을 활성화/비활성화하는 데 사용할 수 있어요.

CREATE SCHEDULED QUERY 문의 기본 동작은 설정 키 hive.scheduled.queries.create.as.enabled로 결정돼요.

해당 일정이 비활성화될 때 진행 중(in-flight)인 예약 실행이 있다면, 이미 실행 중인 실행은 끝까지 마쳐져요. 하지만 더 이상 실행이 트리거되지는 않아요.

Defined AS 구문

[DEFINED] AS

"query"는 실행되도록 예약되는 단일 문(statement) 표현식이에요.

executeSpec 구문

EXECUTE

일정의 다음 실행 시간을 지금(now)으로 변경해요. 디버깅/개발 중에 유용할 수 있어요.

시스템 테이블/뷰 (System tables/views)

예약 쿼리/실행에 대한 정보는 information_schema 또는 sysdb에서 얻을 수 있어요. 권장되는 방법은 information_schema를 사용하는 것이고, sysdb 테이블은 information_schema 레벨 뷰를 만들고 디버깅하기 위해 존재해요.

information_schema.scheduled_queries

다음으로 정의된 예약 쿼리가 있다고 가정해 봐요:

**create scheduled query sc1 cron '0 /10 * * * ? ' as select 1;

information_schema.scheduled_queries 테이블에서 확인해 보면:

select * from information_schema.scheduled_queries;

각 컬럼을 설명하기 위해 결과셋을 전치(transpose)할게요.

| scheduled_query_id | 1 | 내부적으로 모든 예약 쿼리에는 숫자 id가 있음 | | schedule_name | sc1 | 일정 이름 | | enabled | true | 일정이 활성화되어 있으면 true | | cluster_namespace | hive | 이 예약 쿼리가 속한 네임스페이스 | | schedule | 0 */10 * * * ? * | QUARTZ cron 형식으로 기술된 일정 | | user | dev | 쿼리의 소유자/실행자 | | query | select 1 | 예약되는 쿼리 | | next_execution | 2020-01-29 16:50:00 | 기술적 컬럼; 다음 실행이 언제인지 보여줌 |

(schedule_name, cluster_namespace) 는 유일(unique)해요.

information_schema.scheduled_executions

이 뷰는 최근 예약 쿼리 실행에 대한 정보를 얻는 데 사용돼요.

select * from information_schema.scheduled_executions;

이 뷰의 한 레코드는 다음 정보를 담아요:

| scheduled_execution_id | 13 | 모든 예약 쿼리 실행에는 유일한 숫자 id가 있음 | | schedule_name | sc1 | 이 실행이 속한 일정 이름 | | executor_query_id | dev_20200131103008_c9a39b8d-e26b-44cd-b8ae-9d054204dc07 | 실행 엔진이 해당 예약 실행에 부여한 쿼리 id | | state | FINISHED | 실행 상태 | | start_time | 2020-01-31 10:30:06 | 실행 시작 시간 | | end_time | 2020-01-31 10:30:08 | 실행 종료 시간 | | elapsed | 2 | (계산값) end_time-start_time | | error_message | NULL | 쿼리가 FAILED면 오류 메시지가 여기에 표시됨 | | last_update_time | NULL | 실행 중 실행자가 상태 정보를 마지막으로 제공한 갱신 시간 |

실행 상태 (Execution states)

| INITED | 실행자를 배정받은 시점에 예약 실행 레코드가 생성됨. 실행자의 첫 갱신이 올 때까지 INITED 상태 유지 | | EXECUTING | 실행 중 상태의 쿼리는 실행자가 처리 중. 이 단계에서 실행자는 hive.scheduled.queries.executor.progress.report.interval로 정의된 간격으로 쿼리 진행 상황을 보고 | | FAILED | 쿼리 실행이 오류 코드(또는 예외)로 중단됨. 이 상태가 설정되면 error_message도 채워짐 | | FINISHED | 쿼리가 문제없이 완료됨 | | TIMED_OUT | metastore.scheduled.queries.execution.timeout보다 오래 실행 중이면 실행이 타임아웃된 것으로 간주. 예약 쿼리 유지 관리 태스크가 타임아웃된 실행을 확인 |

실행 정보는 얼마나 오래 보관되나요?

예약 쿼리 유지 관리 태스크는 metastore.scheduled.queries.execution.max.age보다 오래된 항목을 제거해요.

설정 (Configuration)

Hive metastore 관련 설정

• metastore.scheduled.queries.enabled (기본값: true) metastore 쪽의 scheduled queries 지원을 제어. 모든 HMS scheduled query 관련 엔드포인트가 오류를 반환하도록 함
• metastore.scheduled.queries.execution.timeout (기본값: 2 minutes) 예약 실행이 최소 이 시간 동안 갱신되지 않으면, 클리너 태스크가 상태를 TIMED_OUT으로 변경
• metastore.scheduled.queries.execution.maint.task.frequency (기본값: 1 minute) 예약 쿼리 유지 관리 태스크의 간격. max age 이상의 실행을 제거하고, 조건이 충족되면 실행을 TIMED_OUT으로 표시
• metastore.scheduled.queries.execution.max.age (기본값: 30 days) 제거되기 전 예약 쿼리 실행 항목의 최대 수명

HiveServer2 관련 설정

• hive.scheduled.queries.executor.enabled (기본값: true) HS2가 예약 쿼리 실행자를 실행할지 제어
• hive.scheduled.queries.namespace (기본값: "hive") 사용할 예약 쿼리 네임스페이스 설정. 새 예약 쿼리는 이 네임스페이스에 생성되고, 실행도 네임스페이스에 바인딩됨
• hive.scheduled.queries.executor.idle.sleep.time (기본값: 1 minute) 예약 쿼리의 존재 여부를 조회하는 사이에 잠드는 시간
• hive.scheduled.queries.executor.progress.report.interval (기본값: 1 minute) 예약 쿼리가 진행 중(in flight)일 때 주기적으로 백그라운드 갱신이 일어나 쿼리 실제 상태를 보고
• hive.scheduled.queries.create.as.enabled (기본값: true) 새로 생성된 예약 쿼리의 기본 동작 설정
• hive.security.authorization.scheduled.queries.supported (기본값: false) 구성된 authorizer가 scheduled query 관련 호출을 처리할 수 있다면 활성화

예제 (Examples)

예제 1 - 일정 사용의 기본 예제

create table t (a integer);

-- 예약 쿼리 생성; 10분마다 새 행 삽입
create scheduled query sc1 cron '0 */10 * * * ? *' as insert into t values (1);
-- hive.scheduled.queries.create.as.enabled에 따라 쿼리가 disabled 모드로 생성될 수 있음
-- 다음으로 활성화할 수 있음:
alter scheduled query sc1 enabled;

-- 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  |
+-----------------------+------------------+------------+----------------------+-------------------+---------+-----------+----------------------+

-- 10분 기다리거나 다음으로 실행:
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 t (a integer);

-- 테이블이 어디 있는지 확인:
desc formatted t;
[...]
| Location:                     | file:/data/hive/warehouse/t                       | NULL                                               |
[...]

-- 터미널에서 테이블 디렉토리에 데이터 로드:
seq 1 10 > /data/hive/warehouse/t/f1

-- hive로 돌아오면 다음을 볼 수 있음
select count(1) from t;
10
-- 그동안 기본 통계는 테이블에 "0"행이 있다고 보여줌
desc formatted t;
[...]
|                               | numRows                                            | 0                                                  |
[...]

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

-- 시간을 기다리거나 다음으로 실행:
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                 |
+---------------------------+------------------+----------------------------------------------------+------------+----------------------+----------------------+------------+------------------+----------------------+

-- 그리고 numrows가 갱신됨
desc formatted t;
[...]
|                               | numRows                                            | 10                                                 |
[...]

-- 더 이상 1분마다 실행하고 싶지 않다면...
alter scheduled query t_analyze disable;

예제 3 - materialized view 재구축

-- 일부 설정... 이미 있을 수 있음
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 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');

-- 데이터 로드
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');

-- 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';

-- mv를 재구축할 일정 생성
create scheduled query mv_rebuild cron '0 */10 * * * ? *' defined as 
  alter materialized view mv1 rebuild;

-- 이 explain에서 mv1이 사용되는 것을 볼 수 있음
EXPLAIN
SELECT empid, deptname FROM emps
JOIN depts ON (emps.deptno = depts.deptno)
WHERE hire_date >= '2018-01-01';

-- 새 레코드 삽입
insert into emps values (1330, 10, 'Bill', 10000, '2020-01-02');

-- 원본 테이블이 스캔됨
EXPLAIN
SELECT empid, deptname FROM emps
JOIN depts ON (emps.deptno = depts.deptno)
WHERE hire_date >= '2018-01-01';

-- 10분 기다리거나 실행
alter scheduled query mv_rebuild execute;

-- 다시 실행... 뷰가 재구축되어야 함
EXPLAIN
SELECT empid, deptname FROM emps
JOIN depts ON (emps.deptno = depts.deptno)
WHERE hire_date >= '2018-01-01';

예제 4 - 수집 (Ingestion)

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

-- 이 테이블이 필터 조건을 id 컬럼에 pushdown 지원하는 외부 테이블이라고 가정
create table s(id integer, cnt integer);

-- 내부 테이블과 offset 테이블 생성
create table t(id integer, cnt integer);
create table t_offset(offset integer);
insert into t_offset values(0);

-- s에 데이터가 추가된다고 가정
insert into s values(1,1);

-- 수집 실행...
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);

-- 10분마다 수집하도록 구성
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);

-- 새 값 추가
insert into s values(2,2),(3,3);

-- 타임아웃이 발생했다고 가정
alter scheduled query ingest execute;

더 알아보기 (Learn more)

• Hive Scheduled Queries 위키에서 자세한 내용을 볼 수 있어요.
• Hive 언어 매뉴얼에서 더 많은 DDL 문법을 확인할 수 있어요.