안녕하세요, 구글 SEO와 사용자 경험 최적화에 진심인 기술 블로그 에디터입니다. 오늘은 Apache Flink의 Python API인 PyFlink를 사용하여 MySQL 데이터베이스에 데이터를 연동하려 할 때 발생하는 py4j.protocol.Py4JJavaError: NoSuchMethodException 에러의 원인과 해결 방법에 대해 깊이 있게 다루어 보겠습니다. 이 문서를 통해 Flink 개발자들이 겪는 흔한 버전 불일치 문제를 명확히 이해하고, 효율적으로 해결할 수 있도록 돕겠습니다.
사용자는 PyFlink 2.2.1 버전(실제로는 Flink 1.x 버전을 의미할 가능성이 높습니다. Flink의 최신 안정 버전은 1.x.x 형태입니다)을 사용하여 간단한 데이터 스트림을 MySQL 데이터베이스에 삽입하는 코드를 작성했습니다. Visual Studio Code 환경에서 PyFlink의 JdbcSink를 활용하여 데이터를 연동하는 과정에서 예외가 발생했습니다.
Traceback (most recent call last):
File "c:\VSCode_Workspace\flink-python\com\aaa\flink\flink-jdbc-test.py", line 34, in <module>
sink = JdbcSink.sink(insert_sql, type_info, jdbc_conn_option, jdbc_exe_option)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\VSCode_Workspace\.venv-etl\Lib\site-packages\pyflink\datastream\connectors\jdbc.py", line 60, in sink
j_builder_method = output_format_clz.getDeclaredMethod('createRowJdbcStatementBuilder',
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\VSCode_Workspace\.venv-etl\Lib\site-packages\py4j\java_gateway.py", line 1322, in __call__
return_value = get_return_value(
^^^^^^^^^^^^^^^^^
File "C:\VSCode_Workspace\.venv-etl\Lib\site-packages\pyflink\util\exceptions.py", line 162, in deco
return f(*a, **kw)
^^^^^^^^^^^
File "C:\VSCode_Workspace\.venv-etl\Lib\site-packages\py4j\protocol.py", line 326, in get_return_value
raise Py4JJavaError(
py4j.protocol.Py4JJavaError: An error occurred while calling o71.getDeclaredMethod.
: java.lang.NoSuchMethodException: org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.createRowJdbcStatementBuilder([I)
at java.base/java.lang.Class.getDeclaredMethod(Class.java:2675)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk/internal/reflect/DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)
at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.base/java.lang.Thread.run(Thread.java:833)
2. 명확한 발생 원인
이 에러의 핵심은 java.lang.NoSuchMethodException: org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.createRowJdbcStatementBuilder([I) 메시지입니다. 이는 PyFlink(Python)가 Py4J를 통해 JVM(Java) 내부의 JdbcOutputFormat 클래스에서 createRowJdbcStatementBuilder라는 이름의 메서드를 찾으려고 시도했지만, int[] (정수 배열) 파라미터를 받는 정확한 시그니처의 메서드를 찾을 수 없었다는 의미입니다.
이러한 NoSuchMethodException이 발생하는 가장 흔하고 명확한 원인은 Apache Flink PyFlink 라이브러리 버전과 Flink JDBC 커넥터 JAR 파일들의 버전 불일치입니다. 세부적으로는 다음과 같습니다.
PyFlink 버전과 Flink 코어 버전: PyFlink 라이브러리는 특정 Flink 코어 버전에 맞춰 빌드됩니다. 예를 들어, pyflink==1.17.1은 Flink 1.17.1 코어에 대응합니다.
커넥터 JAR 버전의 문제: 사용자가 pipeline.jars에 추가한 flink-connector-jdbc-core-4.1.0-2.2.jar 및 flink-connector-jdbc-mysql-4.1.0-2.2.jar 파일의 버전(4.1.0-2.2)은 표준 Flink 버전 명명 규칙(예: 1.17.1)과 일치하지 않습니다. PyFlink 버전이 기대하는 JdbcOutputFormat의 메서드 시그니처와, 실제로 로드된 JDBC 커넥터 JAR 파일에 포함된 JdbcOutputFormat 클래스의 메서드 시그니처가 다르기 때문에 에러가 발생한 것입니다.
메서드 시그니처 변경: Flink 버전이 업데이트됨에 따라 내부 API, 특히 JdbcOutputFormat 같은 핵심 컴포넌트의 메서드 시그니처가 변경되었을 수 있습니다. PyFlink가 오래된 또는 너무 새로운 버전의 커넥터 JAR 파일을 사용하려고 할 때 이러한 불일치가 발생합니다.
결론적으로, PyFlink가 기대하는 메서드 시그니처를 가진 JDBC 커넥터 JAR 파일이 JVM에 로드되지 않아 발생한 문제입니다.
3. 해결 방법 및 코드 예시
이 문제를 해결하는 핵심은 사용 중인 PyFlink 버전에 정확히 일치하는 Flink JDBC Connector JAR 파일들을 사용하는 것입니다. 단계별 해결 방법을 제시합니다.
단계 1: 현재 사용 중인 PyFlink 버전 확인
먼저 Python 환경에서 설치된 pyflink의 버전을 확인해야 합니다.
pip show pyflink
# 또는 Python 코드 내에서
import pyflink
print(pyflink.__version__)
예를 들어, 버전이 1.17.1이라고 가정하겠습니다. (사용자 소스 제목의 “flink 2.2.1″은 Flink 1.17.1 또는 1.18.x 등의 안정 버전을 착각했을 가능성이 높습니다.)
단계 2: PyFlink 버전에 맞는 Flink JDBC Connector JAR 파일 다운로드
확인된 PyFlink 버전에 해당하는 flink-connector-jdbc JAR 파일들을 Apache Flink 공식 다운로드 페이지 또는 Maven Central에서 다운로드합니다.
flink-connector-jdbc_2.12-.jar (core connector)
flink-connector-jdbc-mysql_2.12-.jar (MySQL specific connector)
주의: Scala 버전에 따라 _2.11 또는 _2.12가 붙을 수 있습니다. PyFlink는 일반적으로 Flink의 기본 Scala 버전(보통 _2.12)에 맞춰져 있습니다. PyFlink 1.17.1을 사용한다면, 다음과 같은 JAR 파일들을 다운로드해야 합니다.
MySQL 드라이버 JAR 파일(mysql-connector-j-8.4.0.jar)은 사용해도 무방하지만, Flink와 호환되는 최신 버전을 사용하는 것이 좋습니다. (예: mysql-connector-j-8.0.33.jar 또는 이후 버전).
단계 3: pipeline.jars 경로 업데이트
다운로드한 JAR 파일들을 적절한 디렉토리(예: C:/flink_jars/)에 배치한 후, PyFlink 코드의 config.set_string("pipeline.jars", ...) 부분을 업데이트합니다.
수정된 코드 예시: (PyFlink 1.17.1 기준)
from pyflink.common import Types, Configuration
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.jdbc import JdbcConnectionOptions, JdbcExecutionOptions, JdbcSink
from pyflink.datastream.stream_execution_environment import RuntimeExecutionMode
config = Configuration()
# PyFlink 1.17.1 버전에 맞춰 JAR 파일들을 업데이트합니다.
# 실제 Flink 버전에 맞는 JAR 파일 경로로 수정해주세요.
config.set_string(
"pipeline.jars", "file:///C:/flink_jars/flink-connector-jdbc_2.12-1.17.1.jar;"
"file:///C:/flink_jars/flink-connector-jdbc-mysql_2.12-1.17.1.jar;"
"file:///C:/flink_jars/mysql-connector-j-8.0.33.jar") # MySQL Connector J 드라이버도 호환되는 버전으로 사용
env = StreamExecutionEnvironment.get_execution_environment(config)
env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
type_info = Types.ROW([Types.STRING(), Types.INT()])
ds = env.from_collection([('GM', 300), ('Volvo', 400)], type_info=type_info)
insert_sql = 'insert into Car (brand, price) values (?, ?)'
jdbc_exe_option = JdbcExecutionOptions.builder() \
.with_batch_interval_ms(1000) \
.with_batch_size(200) \
.with_max_retries(5) \
.build()
jdbc_conn_option = JdbcConnectionOptions.JdbcConnectionOptionsBuilder() \
.with_url('jdbc:mysql://localhost:3306/etlmysql') \
.with_driver_name('com.mysql.cj.jdbc.Driver') \
.with_user_name('root') \
.with_password('1234') \
.build()
sink = JdbcSink.sink(insert_sql, type_info, jdbc_conn_option, jdbc_exe_option)
ds.add_sink(sink)
env.execute()
env.close()
이 변경 사항을 적용한 후 코드를 다시 실행하면 NoSuchMethodException 에러가 해결될 것입니다.
4. 향후 예방을 위한 팁
이러한 버전 불일치 문제를 사전에 방지하기 위한 몇 가지 팁입니다.
모든 Flink 컴포넌트 버전 일관성 유지: PyFlink 라이브러리, Flink 클러스터 (만약 사용한다면), 그리고 모든 커넥터 JAR 파일들의 버전을 최대한 일치시키는 것이 중요합니다. Flink 생태계는 빠르게 발전하므로, 버전 간 호환성 문제가 자주 발생할 수 있습니다.
공식 문서 참조: 새로운 Flink 버전을 사용하거나 특정 커넥터를 연동할 때는 항상 Apache Flink 공식 문서를 먼저 참조하여 권장되는 커넥터 버전 및 구성 방법을 확인하세요.
Maven Central 활용: 필요한 JAR 파일들을 다운로드할 때는 Maven Central Repository와 같은 공식 저장소를 활용하여 정확한 버전의 파일을 확보하는 것이 좋습니다.
JDK 버전 확인: Flink는 Java 기반이므로, 사용 중인 JDK 버전이 PyFlink 및 Flink 코어 버전과 호환되는지 확인하는 것도 중요합니다. Flink 1.17+는 Java 8, 11, 17을 지원합니다.
PyFlink 개발 시 버전 관리의 중요성을 항상 인지하고, 문제가 발생했을 때는 관련된 모든 컴포넌트의 버전을 확인하는 습관을 들이시기 바랍니다. 이 가이드가 여러분의 Flink 개발 여정에 도움이 되기를 바랍니다.
![[에러 해결] Python typing.Protocol 함수 시그니처 불일치 원인과 유연한 타입 정의 방법](https://dev-error.com/wp-content/plugins/contextual-related-posts/default.png "[에러 해결] Python typing.Protocol 함수 시그니처 불일치 원인과 유연한 타입 정의 방법")