[Flink] Hive Streaming Source 기능 소개 - unbounded scan

flink 에서 hive 는 unbounded sacn 을 지원한다. 더 정확히 말하면 파티션이나 파일이 생기는걸 주기적으로 감시하다가 데이터를 조회하는 한다는게 더 맞을지도 모르겠다. 이게 뭔 의미가 있나 싶겠지만 스트림 데이터를 다루는것처럼 로직을 flink 에 submit 하면 새로운 파티션이 생길때마다 별도의 스케쥴링 없이 운영이 가능하다는 말이다.

즉, 배치형태로 작업을 구성하지 않고, 스트리밍 데이터를 다루듯 운용할수 있다는 말이다.

 

Hive 데이터를 스트림 데이터처럼 다루기

hive 에 데이터를 적재할때 보통 파티션 단위로 데이터를 적재한다. 

만약, airflow 를 이용해 데이터를 데이터를 재가공하는 로직을 등록한다면 특정 파티션을 센싱하고 있다가 생성이 되면 로직이 실행되는 구조로 DAG를 만드는것이 일반적이다. 이런 접근법은 배치데이터를 구성할때 사용되며 스케쥴러를 이용해서 다음 작업이 실행되고 종료되는 구조를 지니고 있다.

 

그런데, Flink 에서는 kafka 에서 데이터를 consume 하는 스트림데이터를 다루는것처럼 접근할 수 있다.

Flink에서는 배치와 스트림이라는 형태로 접근하지 않고, unbounded 와 bounded 라는 데이터의 경계면이 있는가 없는가로 접근하는데, 이런 접근법의 차이로 배치데이터의 성격을 스트림 스럽게 다루는 접근을 할 수 있었다.

 

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/overview/

어떻게 사용할까?

Hive 에서 t_sample 이라는 테이블에 ymd, hh24 라는 파티션 필드를 가졌고 airflow 를 통해 데이터를 sink 한다고 가정한다. (쉽게말해서 1시간 배치처리) Flink 에서는 Hive 카탈로그 등록을 통해서 쉽게 테이블 조회가 가능한데, 이때 힌트를 넣어서 Stream 데이터처럼 사용할 수 있다.  예를 들면 다음과 같다.

 

아래와 같이 힌트 옵션을 주면, 10분 간격으로 메타스토어를 체크해서, 새로운 파티션이 생길때 데이터를 consume 하는 식으로 대응한다.

-- hive 카탈로그 등록
Flink SQL> CREATE CATALOG  myHive with (
 'type' = 'hive',
 'hive-conf-dir' = '/usr/hdp/2.6.2.0-205/hive/conf'
);

------------------------------------
-- streaming 처럼 데이터를 조회 
-- 10분 주기로 파티션을 체크해서 데이터 생기면 데이터 조회됨
------------------------------------
Flink SQL> select 
  * 
from 
  myHive.sample_db.t_sample
  /*+ OPTIONS(
  'streaming-source.enable' = 'true',
  'streaming-source.partition.include' = 'all',
  'streaming-source.monitor-interval' = '10 min',
  'streaming-source.partition-order' = 'partition-time',
  'partition.time-extractor.kind' = 'default',
  'partition.time-extractor.timestamp-pattern' = '$ymd $hh24:00:00') 
  */
 ;

덤으로, 이 방식은 파티션이 너무 많으면 동작안할수 있음을 인지해야한다.

그 기준은 약 3.2만개 이다. 커보이지만 시간 파티션이라면 3년 반이 지나면 문제가 생기는 수준이다.

이 부분에 대한건 별도의 글로 포스팅 하도록 하겠다.

 

이런 제약은 일단 넘어가고, airflow 나 crotab 같은 스케쥴러 없이, hive 의 데이터를 다룰수 있는건 꽤 재미있는것 같다.