You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Phil Stavridis <ph...@gmail.com> on 2024/05/08 09:50:27 UTC
Flink submit to cluster issue
I am running a Flink job locally using
python -m job.py
and it runs fine.
The job is:
calcul_count = t_env.execute_sql("""
SELECT
username,
COUNT(action) AS a_count
FROM kafka_logs
GROUP BY username
""")
with calcul_count.collect() as results:
for row in results:
print(row)
When i try to submit it to the Flink JobManager running as a Docker service with:
docker-compose exec -e KAFKA_BOOTSTRAP_SERVERS="kafka:9092" jobmanager flink run -py /opt/app/job.py
I am getting error:
for row in results:
File "/opt/flink/opt/python/pyflink.zip/pyflink/table/table_result.py", line 240, in __next__
File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__
File "/opt/flink/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 146, in deco
File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o176.hasNext.
: java.lang.RuntimeException: Failed to fetch next result
Caused by: java.io.IOException: Failed to fetch job execution result
Caused by: java.util.concurrent.ExecutionException: org.apache.flink.client.program.ProgramInvocationException: Job failed
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: class org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArrayDeserializer is not an instance of org.apache.kafka.common.serialization.Deserializer
This is my flink cluster Dockerfile:
FROM flink:1.18.1-scala_2.12
RUN apt-get update && \
apt-get install -y python3 python3-pip && \
ln -s /usr/bin/python3 /usr/bin/python && \
apt-get clean && \
rm -rf /var/lib/apt/lists/*
RUN pip install apache-flink==1.18.1
# Download the necessary Flink connector jars
ADD https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.1.0-1.18/flink-sql-connector-kafka-3.1.0-1.18.jar /opt/flink/lib/
ADD https://repo.maven.apache.org/maven2/org/apache/flink/flink-json/1.18.1/flink-json-1.18.1.jar /opt/flink/lib/
WORKDIR /opt/flink
Is it a library missing from the Flink cluster? What would cause that error?
I have also tried to run the job.py by submitting it on a Flink cluster on EMR which has the same version of Flink as my local Flink cluster which runs as a Docker container with:
/usr/lib/flink/bin/flink run -py /opt/app/job.py
after copying the job on the Flink JobManager on EMR but i am getting error:
calcul_count = t_env.execute_sql("""
File "/usr/lib/flink/opt/python/pyflink.zip/pyflink/table/table_environment.py", line 837, in execute_sql
File "/usr/lib/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__
File "/usr/lib/flink/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 146, in deco
File "/usr/lib/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o16.executeSql.
: java.lang.NoSuchFieldError: SPOT_INSTANCE_INTERRUPTION_NOTICE_DURATION
at org.apache.flink.streaming.api.environment.CheckpointConfig.setSpotInstanceInterruptionNoticeDuration(CheckpointConfig.java:891)
at org.apache.flink.streaming.api.graph.StreamGraph.setSpotInstanceInterruptionNoticeDuration(StreamGraph.java:182)
Re: Flink submit to cluster issue
Posted by Biao Geng <bi...@gmail.com>.
Hi Phil,
>>> Is it a library missing from the Flink cluster? What would cause that
error?
In most cases, it implies the collision of jar dependencies.
flink-sql-connector-kafka jar should be enough to run the job and you may
need to check your classpath in the jobmanager to see if any other jars
that contain kafka are included.
Another hacking method is to set config 'classloader.resolve-order' to
'parent-first' in your flink-conf.yaml and restart your flink cluster.
>>>> java.lang.NoSuchFieldError: SPOT_INSTANCE_INTERRUPTION_NOTICE_DURATION
The method name SPOT_INSTANCE_INTERRUPTION_NOTICE_DURATION does not exist
in flink repo. Maybe it is better to contact the AWS expert to find the
cause of this problem.
Best,
Biao Geng
Phil Stavridis <ph...@gmail.com> 于2024年5月8日周三 17:51写道：
> I am running a Flink job locally using
>
> python -m job.py
>
> and it runs fine.
>
> The job is:
>
> calcul_count = t_env.execute_sql("""
> SELECT
> username,
> COUNT(action) AS a_count
> FROM kafka_logs
> GROUP BY username
> """)
>
> with calcul_count.collect() as results:
> for row in results:
> print(row)
>
>
> When i try to submit it to the Flink JobManager running as a Docker
> service with:
>
> docker-compose exec -e KAFKA_BOOTSTRAP_SERVERS="kafka:9092" jobmanager
> flink run -py /opt/app/job.py
>
>
> I am getting error:
>
> for row in results:
> File "/opt/flink/opt/python/pyflink.zip/pyflink/table/table_result.py",
> line 240, in __next__
> File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py",
> line 1322, in __call__
> File "/opt/flink/opt/python/pyflink.zip/pyflink/util/exceptions.py", line
> 146, in deco
> File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/protocol.py", line
> 326, in get_return_value
> py4j.protocol.Py4JJavaError: An error occurred while calling o176.hasNext.
> : java.lang.RuntimeException: Failed to fetch next result
>
> Caused by: java.io.IOException: Failed to fetch job execution result
> Caused by: java.util.concurrent.ExecutionException:
> org.apache.flink.client.program.ProgramInvocationException: Job failed
> Caused by: org.apache.kafka.common.KafkaException:
> org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
> Caused by: org.apache.kafka.common.KafkaException:
> org.apache.kafka.common.KafkaException: class
> org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArrayDeserializer
> is not an instance of org.apache.kafka.common.serialization.Deserializer
>
>
> This is my flink cluster Dockerfile:
>
> FROM flink:1.18.1-scala_2.12
>
> RUN apt-get update && \
> apt-get install -y python3 python3-pip && \
> ln -s /usr/bin/python3 /usr/bin/python && \
> apt-get clean && \
> rm -rf /var/lib/apt/lists/*
>
> RUN pip install apache-flink==1.18.1
>
> # Download the necessary Flink connector jars
> ADD
> https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.1.0-1.18/flink-sql-connector-kafka-3.1.0-1.18.jar
> /opt/flink/lib/
> ADD
> https://repo.maven.apache.org/maven2/org/apache/flink/flink-json/1.18.1/flink-json-1.18.1.jar
> /opt/flink/lib/
>
> WORKDIR /opt/flink
> Is it a library missing from the Flink cluster? What would cause that
> error?
>
>
> I have also tried to run the job.py by submitting it on a Flink cluster on
> EMR which has the same version of Flink as my local Flink cluster which
> runs as a Docker container with:
>
> /usr/lib/flink/bin/flink run -py /opt/app/job.py
> after copying the job on the Flink JobManager on EMR but i am getting
> error:
>
> calcul_count = t_env.execute_sql("""
> File
> "/usr/lib/flink/opt/python/pyflink.zip/pyflink/table/table_environment.py",
> line 837, in execute_sql
> File
> "/usr/lib/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py",
> line 1322, in __call__
> File "/usr/lib/flink/opt/python/pyflink.zip/pyflink/util/exceptions.py",
> line 146, in deco
> File "/usr/lib/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/protocol.py",
> line 326, in get_return_value
> py4j.protocol.Py4JJavaError: An error occurred while calling
> o16.executeSql.
> : java.lang.NoSuchFieldError: SPOT_INSTANCE_INTERRUPTION_NOTICE_DURATION
> at
> org.apache.flink.streaming.api.environment.CheckpointConfig.setSpotInstanceInterruptionNoticeDuration(CheckpointConfig.java:891)
> at
> org.apache.flink.streaming.api.graph.StreamGraph.setSpotInstanceInterruptionNoticeDuration(StreamGraph.java:182)
>