https://flink.apache.org/ logo
Join Slack
Powered by
Hey using this code to read from kafka using kafka...
# troubleshooting
k
Hey using this code to read from kafka using kafkajsonformat.py
Copy code
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.typeinfo import Types
from pyflink.common.watermark_strategy import WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaSink, KafkaOffsetsInitializer, KafkaRecordSerializationSchema, DeliveryGuarantee


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

    brokers = "cluster-0-kafka-bootstrap.strimzi.svc.cluster.local:9092"
    source = KafkaSource.builder() \
        .set_bootstrap_servers(brokers) \
        .set_topics("locations") \
        .set_group_id("flinktestpers") \
        .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
        .set_value_only_deserializer(SimpleStringSchema()) \
        .build()

    ds = env.from_source(source, WatermarkStrategy.no_watermarks(), "Kafka Source")

    # Apply the FlatMap transformation
    ds_flatmapped = ds.flat_map(lambda value: value.split())

    # Apply the custom log processing function
    processed_log = ds_flatmapped.map(lambda word: word.upper())

    # Define the Kafka sink
    sink = KafkaSink.builder() \
            .set_bootstrap_servers(brokers) \
            .set_record_serializer(
                KafkaRecordSerializationSchema.builder()
                    .set_topic("flinksink")  # Set your output topic here
                    .set_value_serialization_schema(SimpleStringSchema())
                    .build()
            ) \
            .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE) \
            .build()

    processed_log.sink_to(sink)

    env.execute('kafkaread')


if __name__ == '__main__':
    kafkaread()
and flinkjob.yaml deployment
Copy code
apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkDeployment
metadata:
  name: kafkatest
spec:
  image: flink-job:latest
  flinkVersion: v1_16
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
  serviceAccount: flink
  jobManager:
    resource:
      memory: "3048m"
      cpu: 2
  taskManager:
    resource:
      memory: "3048m"
      cpu: 2
  job:
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args: ["-pyclientexec", "/usr/local/bin/python3", "-py", "/opt/flink/lib/kafkajsonformat.py"]
    parallelism: 2
    upgradeMode: stateless
error
Copy code
2023-07-28 20:01:42,241 INFO  org.apache.flink.python.env.AbstractPythonEnvironmentManager [] - PYTHONPATH of python worker: null
2023-07-28 20:01:42,243 INFO  org.apache.flink.python.env.AbstractPythonEnvironmentManager [] - Python interpreter path: python
2023-07-28 20:01:42,351 WARN  org.apache.flink.streaming.api.operators.AbstractStreamOperator [] - Failed to clean up the leaking objects.
java.lang.ClassCastException: class java.io.ObjectStreamClass$Caches$1 cannot be cast to class java.util.Map (java.io.ObjectStreamClass$Caches$1 and java.util.Map are in module java.base of loader 'bootstrap')
	at org.apache.flink.streaming.api.utils.ClassLeakCleaner.clearCache(ClassLeakCleaner.java:58) ~[flink-python-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.api.utils.ClassLeakCleaner.cleanUpLeakingClasses(ClassLeakCleaner.java:39) ~[flink-python-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.close(AbstractPythonFunctionOperator.java:142) ~[flink-python-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.api.operators.python.process.AbstractExternalPythonFunctionOperator.close(AbstractExternalPythonFunctionOperator.java:74) ~[flink-python-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:163) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.closeAllOperators(RegularOperatorChain.java:125) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.closeAllOperators(StreamTask.java:1030) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.core.fs.AutoCloseableRegistry.doClose(AutoCloseableRegistry.java:72) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.util.AbstractAutoCloseableRegistry.close(AbstractAutoCloseableRegistry.java:127) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.cleanUp(StreamTask.java:948) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.runtime.taskmanager.Task.lambda$restoreAndInvoke$0(Task.java:917) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935) [flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:917) [flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728) [flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550) [flink-dist-1.16.2.jar:1.16.2]
	at java.lang.Thread.run(Unknown Source) [?:?]
I have connector downloaded in docker image /opt/flink/lib for kafka do I need to provide path of kafka or I am missing something ??
d
@kapil singh Hey, this is an known issue which should have already been fixed in 1.16.1. It seems that you are using 1.16.0: 
flink-python-1.16.0.jar
, see https://issues.apache.org/jira/browse/FLINK-30308 for more details.
k
Hey I started using 16.2 and facing this issue now
Copy code
- Source: Kafka Source -> FlatMap, Map -> Sink: Writer -> Sink: Committer (2/2)#0 (fcc3d39aa408cd4e8448992b10e14ade_cbc357ccb763df2852fee8c4fc7d55f2_1_0) switched from INITIALIZING to FAILED with failure cause: java.lang.NoClassDefFoundError: Could not initialize class org.apache.beam.sdk.options.PipelineOptionsFactory
	at org.apache.flink.streaming.api.runners.python.beam.BeamPythonFunctionRunner.open(BeamPythonFunctionRunner.java:247)
	at org.apache.flink.streaming.api.operators.python.process.AbstractExternalPythonFunctionOperator.open(AbstractExternalPythonFunctionOperator.java:57)
	at org.apache.flink.streaming.api.operators.python.process.AbstractExternalDataStreamPythonFunctionOperator.open(AbstractExternalDataStreamPythonFunctionOperator.java:85)
	at org.apache.flink.streaming.api.operators.python.process.AbstractExternalOneInputPythonFunctionOperator.open(AbstractExternalOneInputPythonFunctionOperator.java:117)
	at org.apache.flink.streaming.api.operators.python.process.ExternalPythonProcessOperator.open(ExternalPythonProcessOperator.java:64)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:107)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:731)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:706)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:672)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:904)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.lang.Thread.run(Thread.java:750)

2023-08-01 07:17:06,722 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Source: Kafka Source -> FlatMap, Map -> Sink: Writer -> Sink: Committer (1/2)#0 (fcc3d39aa408cd4e8448992b10e14ade_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from INITIALIZING to FAILED with failure cause: java.lang.NoClassDefFoundError: com/fasterxml/jackson/core/Versioned
	at java.lang.ClassLoader.defineClass1(Native Method)
	at java.lang.ClassLoader.defineClass(ClassLoader.java:756)
	at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
	at java.net.URLClassLoader.defineClass(URLClassLoader.java:473)
	at java.net.URLClassLoader.access$100(URLClassLoader.java:74)
	at java.net.URLClassLoader$1.run(URLClassLoader.java:369)
	at java.net.URLClassLoader$1.run(URLClassLoader.java:363)
	at java.security.AccessController.doPrivileged(Native Method)
	at java.net.URLClassLoader.findClass(URLClassLoader.java:362)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
	at org.apache.beam.sdk.options.PipelineOptionsFactory.<clinit>(PipelineOptionsFactory.java:476)
	at org.apache.flink.streaming.api.runners.python.beam.BeamPythonFunctionRunner.open(BeamPythonFunctionRunner.java:247)
	at org.apache.flink.streaming.api.operators.python.process.AbstractExternalPythonFunctionOperator.open(AbstractExternalPythonFunctionOperator.java:57)
	at org.apache.flink.streaming.api.operators.python.process.AbstractExternalDataStreamPythonFunctionOperator.open(AbstractExternalDataStreamPythonFunctionOperator.java:85)
	at org.apache.flink.streaming.api.operators.python.process.AbstractExternalOneInputPythonFunctionOperator.open(AbstractExternalOneInputPythonFunctionOperator.java:117)
	at org.apache.flink.streaming.api.operators.python.process.ExternalPythonProcessOperator.open(ExternalPythonProcessOperator.java:64)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:107)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:731)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:706)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:672)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:904)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.ClassNotFoundException: com.fasterxml.jackson.core.Versioned
	at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
	... 28 more
2 Views
Open in Slack
PreviousNext
Powered by