Hi everyone, I’m trying to integrate Elasticsearch 7 with PyFlink (version 1.20.1) using the jar

flink-connector-elasticsearch7-3.1.0-1.20.jar

. when I run my job, I’m getting the following error:

Exception in thread "Thread-5" java.lang.NoClassDefFoundError: org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBuilderBase
py4j.protocol.Py4JError: org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder does not exist in the JVM

here’s a minimal example of my code

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.elasticsearch import Elasticsearch7SinkBuilder, ElasticsearchEmitter
import json

env = StreamExecutionEnvironment.get_execution_environment()
env.add_jars("file:///C:/flink-1.20.1/lib/flink-connector-elasticsearch7-3.1.0-1.20.jar")

data_stream = env.from_collection([
    {"id": "1", "name": "John", "age": 25},
    {"id": "2", "name": "Jane", "age": 30}
])

json_stream = data_stream.map(lambda x: json.dumps(x))

es7_sink = Elasticsearch7SinkBuilder() \
    .set_bulk_flush_max_actions(1) \
    .set_emitter(ElasticsearchEmitter.static_index('my-index', 'id')) \
    .set_hosts(['10.0.0.102:9200']) \
    .build()

json_stream.sink_to(es7_sink)
env.execute("Flink Elasticsearch 7 Job")

i’m running it using

flink run -py sampletestelastic.py

has anyone faced this issue before? seems like

ElasticsearchSinkBuilderBase

class is missing from the jar or not visible to PyFlink. do i need an extra dependency or different jar for flink 1.20.1? It seems like

ElasticsearchSinkBuilderBase

might be missing or not accessible from the JVM side. any guidance or suggestions would be really appreciated

I'm guessing you are missing transitive dependencies. See https://mvnrepository.com/artifact/org.apache.flink/flink-connector-elasticsearch7/3.1.0-1.20 You'll also need to the the jars for: • flink-connector-elasticsearch-base • httpcore-nio • elasticsearch-rest-high-level-client

Have you checked if the required class is present on your jar file?

ElasticsearchSinkBuilderBase

is part of the flink-connector-elasticsearch-base, which was not added (at least not in the given example code). Only flink-connector-elasticsearch was added with env.add_jars in the example

Hi @Andreas Bube and @Pedro Mázala Thanks in advance! As you suggested, I’ve added the base JAR and also included the other JARs you mentioned in the Flink lib directory. However, when I try to execute my PyFlink job, I’m facing the following issue with the

set_hosts

method: I’m using the following setup: • Flink version: 1.20.1 • Python version: 3.10 • Java version: 11 Could you please help me understand if I’m missing any dependency or if there’s a version compatibility issue between the Elasticsearch connector and Flink?

Traceback (most recent call last):
  File "/mnt/d/A.VASUGI/New_folder_flink_git/sampletestelastic.py", line 20, in <module>
    .set_hosts(['<http://10.0.0.102:9200>']) \
  File "/mnt/c/flink-1.20.1/opt/python/pyflink.zip/pyflink/datastream/connectors/elasticsearch.py", line 131, in set_hosts
  File "/mnt/c/flink-1.20.1/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__
  File "/mnt/c/flink-1.20.1/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 146, in deco
  File "/mnt/c/flink-1.20.1/opt/python/py4j-0.10.9.7-src.zip/py4j/protocol.py", line 330, in get_return_value
py4j.protocol.Py4JError: An error occurred while calling o67.setHosts. Trace:
org.apache.flink.api.python.shaded.py4j.Py4JException: Method setHosts([class [Lorg.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost;]) does not exist

I have no clue. I've never used the elasticsearch connector 🤷 You obviously would have to make sure you added compatible versions of the three JAR files. Can you list the JAR files you added along with their versions and a snippet of code showing how you added them?

Hi @Andreas Bube Below are the dependencies I used, along with the corresponding code that I tried to execute.

Oh! I just spotted you're adding the sql connector but relying on the datastream one. Try using flink-connector-elasticsearch7-3.1.0-1.20.jar instead

Apologies for the inconvenience, but I’m still encountering the same issue even after removing and re-adding the correct JAR file path.

In the example you've provided you have only added two JAR files. You'll need to add all of the dependencies, including their transitive ones.