Flink 1.15 + Hive 충돌해결 - org.apache.flink.table.planner.delegation.ParserFactory

Flink 1.14 에서는 잘 동작하던 쿼리가 Flink 1.15 에서 오류가 발생되었다.

쿼리는 kafka 테이블을 mysql 혹은 hdfs 에 sink 하는 로직이었고, 1.15.2 버전이 나와서 그걸 써도 동일한 문제가 발생되었다.

Flink SQL>
> INSERT INTO MysqlSource
> SELECT
>   DATE_FORMAT(window_start, 'yyyy-MM-dd') AS ymd,
>   DATE_FORMAT(window_start, 'HHmm') AS hh24,
>   keyword,
>   COUNT(*) as cnt
> FROM TABLE(
>   TUMBLE(
>     DATA => TABLE KafkaSource,
>     TIMECOL => DESCRIPTOR(logTime),
>     SIZE => INTERVAL '1' HOUR ))
[INFO] Submitting SQL update statement to the cluster...
[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: org.apache.flink.table.planner.delegation.ParserFactory

오류 원인과 문제

우선 오류가 재현되는 케이스는 flink 1.15 + Hive Connector 모듈을 같이 구성한 패키지에서만 오류가 발생되었다.

오류메시지를 다시 확인해 보면 "org.apache.flink.table.planner.delegation.ParserFactory" 관련 클래스에서 영향을 받았는데 사실 1.15 릴리즈노트를 확인해보면 알겠지만 꽤 많은 부분에서 변경이 일어났음을 알수 있다.

 

해당 릴리즈노트의 이슈와 연관된 이슈를 찾다보니 FLINK-27020 이슈가 명시적이고 유사한 케이스임을 확인했다.

hive 모듈을 로딩후 쿼리를 hive 스타일로 바꾸려고 하는데 실패

참고로 오류가 안났던 flink 1.14 에서는 오류없이 정상적으로 동작했다.

 

해결방법

해당 이슈에도 팁이 있고, 릴리즈노트에도 팁이 존재했다. opt 에 있는 파일과 table-planner 를 바꿔치면 하위호환성을 유지할수 있다는 내용이다. 이런내용을 hive connector 에 언급해줘야지 잘 보지도 않는 곳에 언급되어 있다니 시간을 너무 낭비한 느낌이크다.

이미 릴리즈 노트에는 하위호환성을 위한 팁 설명이 있었다.

실제로 ./opt/flink-table-planner_2.12-1.15.0 과 ./lib/flink-table-planner-loader-1.15.jar 를 파일을 바꿔치기하면 Hive 모듈을 로딩후, hive 스타일의 쿼리로 표현하도록 세팅해도 오류가 발생되지 않았다.

## FLINK_HOME 은 지정되어있다고 가정
$ cd $FLINK_HOME
$ cp ./opt/flink-table-planner_2.12-1.15.0.jar ./lib
$ mv ./lib/flink-table-planner-loader-1.15.0.jar ./opt
$ ./bin/sql-client.sh 

Flink SQL> load module hive;
[INFO] Execute statement succeed.

Flink SQL> use modules hive,core;
[INFO] Execute statement succeed.

Flink SQL> set table.sql-dialect=hive;
[INFO] Session property has been set. <-- 오류없음

그리고, 실패나던 쿼리도 위와 같이 jar 를 바꿔치고나니 "java.lang.ClassNotFoundException: org.apache.flink.table.planner.delegation.ParserFactory" 오류없이 쿼리 제출에 성공했다.

 

단순히 hive connector 관련 jar 를 lib 폴더에 포함되어 있기만 하고, kafka to hdfs 같은 로직에서도 비슷한 오류가 나서 찾는데 더 시간이 걸리긴했는데, hive dialect 관련 문서에 관련내용이 추가되었다.

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/hive/hive_dialect/#notice

Hive Dialect