Hi, i am trying to connect to kafka locally with that code:

from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
from pyflink.table import StreamTableEnvironment
from pyflink.common import WatermarkStrategy
from pyflink.datastream.formats.json import JsonRowDeserializationSchema
from pyflink.table import DataTypes
from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer, FlinkKafkaProducer, KafkaSource, KafkaOffsetsInitializer, KafkaSink, KafkaRecordSerializationSchema
from pyflink.common.typeinfo import Types
from pyflink.table.udf import udf
from pyflink.table.expressions import col
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.connectors import FlinkKafkaConsumer


env = StreamExecutionEnvironment.get_execution_environment()
env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
env.set_parallelism(2)

t_env = StreamTableEnvironment.create(env)

source = KafkaSource.builder() \
    .set_bootstrap_servers("localhost:9092") \
    .set_topics("quicktart") \
    .set_group_id("myGroup") \
    .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
    .set_value_only_deserializer(
        JsonRowDeserializationSchema.builder()
        .type_info(Types.ROW([Types.STRING(), Types.STRING(),Types.STRING(), Types.STRING(),Types.FLOAT(), Types.STRING()]))
        .build()) \
    .build()

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

and i am getting this error:

TypeError: Could not found the Java class 'org.apache.flink.connector.kafka.source.KafkaSource.builder'. The Java dependencies could be specified via command line argument '--jarfile' or the config option 'pipeline.jars'

Any help would be appreciated!

Problem is you are not setting up the jar as mentioned in the docs. You need to use something like

env.add_jars("path")