[FLINK] yarn 세션 모드 : 클러스터 N대로 sql-client 접속하는 방법

Flink 에서 세션모드로 application 이나 sql-client.sh 를 사용하려고 했는데, 이런 오류가 발생했다.

참고로 N대의 세션 클러스터를 기동하기위해 주키퍼 노드 관련 옵션을 추가해서 기동했다. 

다음과 같이 기동하면 2개의 컨테이너가 유지되고, 따로 클러스터 유지가 가능하다.

참고로 -z 옵션은 "high-availability.cluster-id" 설정과 연계되는데 이게 기본적으로 yarn 의 application id 로 지정되다보니 데몬은 띄웠는데 어플리케이션이나 sql-client.sh 에서 쿼리 실행할 때 문제가  있다.

 ./bin/yarn-session.sh --detached -z yarn-flink-001 -nm myFlink01
 ./bin/yarn-session.sh --detached -z yarn-flink-002 -nm myFlink02

 

특정 클러스터에 쿼리 요청하기

위와 같이 2개의 세션모드의 클러스터를 기동했다고 가정했을때, 특정 클러스터에 요청하기위해서는 두가지 정보를 알아야 한다.

 

•  클러스터 아이디 (-z 옵션 줬던값)
  • 예시 : yarn-flink-001 , yarn-flink-002
• yarn application id 
  
  • 예시 : application_1599447873137_1590874 ,application_1599447873137_1590875 

만약, 클러스터 아이디가 "yarn-flink-001" 이고, yarn application id 가 "application_1599447873137_1590874" 인 클러스터에 쿼리를 날리고 싶다면 다음과 같이 set 명령이 선행되어야 한다.

Flink SQL> set high-availability.cluster-id=yarn-flink-001;
[INFO] Session property has been set.

Flink SQL> set yarn.application.id=application_1599447873137_1590874;

Flink SQL> select * from sample_table limit 4;

만약 지정없이 실행했다면?

"high-availability.cluster-id" 와 "yarn.application.id" 를 지정하지 않는다면, 아래와 같은 오류가 반갑게 맞이 했을것이다. 그 이유는 flink-conf.yaml 설정에 "high-availability.cluster-id" 값을 지정하거나, yarn-session.sh 에서 -z 옵션으로 클러스터 아이디를 지정했어도, sql-client 에서 기본값은 yarn 의 application id 로 세팅되기 때문에 정보가 맞지 않아서 쿼리를 돌리지 못하는 문제가 발생된다.

 

이 문제는 인터넷을 아무리 뒤져도 잘 안나와서 꽤 오래 삽질했는데, 이건 application 을 yarn-session 모드로 실행할때 동일하므로 꼭 알아두자.

Caused by: org.apache.flink.runtime.client.JobSubmissionException: Failed to submit JobGraph.
        at org.apache.flink.client.program.rest.RestClusterClient.lambda$submitJob$11(RestClusterClient.java:433)
        at java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:870)
        at java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:852)
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
        at org.apache.flink.util.concurrent.FutureUtils.lambda$retryOperationWithDelay$9(FutureUtils.java:373)
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
        at org.apache.flink.util.concurrent.FutureUtils$Timeout.run(FutureUtils.java:1237)
        at org.apache.flink.util.concurrent.DirectExecutorService.execute(DirectExecutorService.java:217)
        at org.apache.flink.util.concurrent.FutureUtils.lambda$orTimeout$15(FutureUtils.java:591)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)