[오류] Flink SQL 에서 HIVE 데이터를 조회 못하는 문제

Flink 에서 Hive Connector 를 연동해서, 카탈로그에 등록하면 hive 의 데이터를 조회해볼 수 있다.

기본적으로는 flink 의 taskManager 의 슬롯안에서 데이터를 처리하기 때문에 (즉, M/R 이나 TEZ 엔진이 도는 yarn 기반아닌)

여유있는 슬롯보다 더 큰 슬롯요청을 하게되면 조회가 안되는 상황이 발생될 수 있다.

 

현상 : flink에서 hive 테이블 조회가 안되는 현상

결론부터 말하면, 나같은 경우는 hive 테이블을 조회하기 위해서 taskManager 의 슬롯이 50개나 요청되었는데, 실제 가용가능한 슬롯은 2개밖에 없어서 나타난 현상이었다. 명시적으로 쿼리 에러가 났으면 빨리 찾았을텐데...

다음과 같이 쿼리 요청했는데 기냥 대기상태로 있고 에러가 바로 나타나지 않았다. 이럴땐 flink 의 대시보드를 확인해 봐야한다. 

-- 카탈로그 등록 / 이동 (하이브의 메타스토어 등록)
Flink SQL> create catalog myHive with (
>  'type' = 'hive',
>  'hive-conf-dir' = '/home/myUser/apache-hive-2.3.6-bin/conf'
> );

Flink SQL> use catalog myHive;
Flink SQL> set sql-client.execution.result-mode=tableau; -- 조회할때 ui 형태가 아닌 text 결과로 보기위해서

-- 결과가 안나오고 멍때림
Flink SQL> select count(*) from fact.t_order where ymd = '2021-10-14' ;
+----+----------------------+
| op |               EXPR$0 |
+----+----------------------+

다음과 같이 flink 대시보드를 확인하면 리소스가 부족해서 실패나면서 restarting 하면서 무한 재시도 중이었고

요청된 tasks 는 50개였는데, 여유 슬롯은 2개인가 5개 밖에 없어서 계속 실패나는 상황이었다.

해결방법 : hive source 리소스 제한

flink의 여유 슬롯은 적은데 개념없게(?) 50개가 슬롯을 요청해서 나타난 현상을 해결하려면 taskManager 를 더 띄워서 가용가능한 슬롯을 늘려주는 방법이 있는데, 사실 테스트를 하기위해 띄워놓은거라 무한정 띄워놓을순 없었기 때문에 리소스를 제한하기로 했다.

 

Hive 의 테이블을 읽어낼때 병렬화 갯수를 제한하는 "table.exec.hive.infer-source-parallelism.max" 값을 지정해서 hive 의 source parallelism 을 제한해 줄 수 있다. 참고로  관련된 flink 문서 링크는 다음과 같다.

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/hive/hive_read_write/#source-parallelism-inference

 

설정을 적용하면 다음과 같이 멍때리던(?) 쿼리가 정상적으로 잘 동작함을 알 수 있다.

----------------------------------------------------------------
-- [문제] flink 여유슬롯이 부족해서 멍때리고 있음 (대시보드에서 실패난거 확인가능)
----------------------------------------------------------------
Flink SQL> select count(*) from fact.t_order where ymd = '2021-10-14' ;
+----+----------------------+
| op |               EXPR$0 |
+----+----------------------+

----------------------------------------------------------------
-- [해결] hive source 의 병렬수의 상한을 정하면 해결됨
----------------------------------------------------------------
Flink SQL> set table.exec.hive.infer-source-parallelism.max=2; -- 가용가능한 슬롯보다 낮게 설정
Flink SQL> select count(*) from fact.t_order where ymd = '2021-10-14'
+----+----------------------+
| op |               EXPR$0 |
+----+----------------------+
| +I |                    1 |
| -U |                    1 |
| +U |                    2 |
     ..변경되는 값이 쭉 보여짐...
| -U |                  245 |
| +U |                  246 |
+----+----------------------+
Received a total of 491 rows