Hello I m trying to get pyflink running in docker with kafka Apache Flink #troubleshooting

Hello, I'm trying to get pyflink running in docker...

Pavel Jezek

08/09/2023, 12:02 PM

Hello, I'm trying to get pyflink running in docker with kafka but I'm receiving errors when trying to submit python job to flink.

Copy code

py4j.protocol.Py4JJavaError: An error occurred while calling None.org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.
: java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArrayDeserializer

Here is my docker-compose.yml:

Copy code

version: '3'

services:

  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: <PLAINTEXT://kafka:9092>
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - 9092:9092
  jobmanager:
    build: .
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager

  taskmanager:
    build: .
    depends_on:
      - jobmanager
    command: taskmanager
    links:
      - "jobmanager:jobmanager"
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager

  kafka-cli:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - kafka
    command: "bash -c 'while true; do sleep 1000; done'"

  kafka-setup:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - kafka
    volumes:
      - ./kafka-setup.sh:/kafka-setup.sh
    command: ["bash", "/kafka-setup.sh"]

And Dockerfile:

Copy code

FROM --platform=linux/amd64 flink:1.17.1

# Install Python 3 and pip
RUN apt-get update -y && \
apt-get install -y python3 python3-pip python3-dev && rm -rf /var/lib/apt/lists/*
RUN ln -s /usr/bin/python3 /usr/bin/python

RUN pip3 install apache-flink==1.17.1

# Install Kafka connector dependencies
RUN mkdir -p /opt/flink/usrlib
RUN wget -P /opt/flink/usrlib <https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-kafka/1.17.1/flink-connector-kafka-1.17.1.jar>
RUN wget -P /opt/flink/usrlib <https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-base/1.17.1/flink-connector-base-1.17.1.jar>
RUN wget -P /opt/flink/usrlib <https://repo.maven.apache.org/maven2/org/apache/kafka/kafka-clients/3.2.3/kafka-clients-3.2.3.jar>

RUN echo "pipeline.jars: file:///opt/flink/usrlib/flink-connector-kafka-1.17.1.jar;file:///opt/flink/usrlib/flink-connector-base-1.17.1.jar;file:///opt/flink/usrlib/kafka-clients-3.2.3.jar" >> /opt/flink/conf/flink-conf.yaml

# Copy the Python files
COPY main.py /opt/flink/usrlib/main.py

And contents of python script:

Copy code

from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer

def main():
    env = StreamExecutionEnvironment.get_execution_environment()

    kafka_props = {"bootstrap.servers": "kafka:9092", "group.id": "flink_group"}

    kafka_consumer = FlinkKafkaConsumer("input_topic", SimpleStringSchema(), kafka_props)
    kafka_producer = FlinkKafkaProducer("output_topic", SimpleStringSchema(), kafka_props)

    stream = env.add_source(kafka_consumer)

    stream.map(lambda x: x.upper(), output_type=Types.STRING()) \
        .add_sink(kafka_producer)

    env.execute("Kafka Flink Job")

if __name__ == "__main__":
    main()

And here is how I'm trying to submit the job:

Copy code

docker exec -it flink-infra-poc-jobmanager-1 ./bin/flink run -py /opt/flink/usrlib/main.py --jarfile /opt/flink/usrlib/flink-connector-kafka-1.17.1.jar --jarfile /opt/flink/usrlib/kafka-clients-3.2.3.jar --jarfile /opt/flink/usrlib/flink-connector-kafka-1.17.1.jar

If I don't include the