Error when using pyflink on databricks, An error occurred while trying to connect to the Java server

[witnessthee]

Hi, right now I am trying to run a pyflink script that can connect to a kafka server. When I run that script, I got an error "An error occurred while trying to connect to the Java server 127.0.0.1:35529". Do I need to install a extra jdk for that?

error message:

Py4JNetworkError: An error occurred while trying to connect to the Java server (127.0.0.1:35529)

---------------------------------------------------------------------------

IndexError Traceback (most recent call last)

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/py4j/java_gateway.py in _get_connection(self)

976 try:

--> 977 connection = self.deque.pop()

978 except IndexError:

IndexError: pop from an empty deque

During handling of the above exception, another exception occurred:

ConnectionRefusedError Traceback (most recent call last)

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/py4j/java_gateway.py in start(self)

1114 try:

-> 1115 self.socket.connect((self.address, self.port))

1116 self.stream = self.socket.makefile("rb")

ConnectionRefusedError: [Errno 111] Connection refused

During handling of the above exception, another exception occurred:

Py4JNetworkError Traceback (most recent call last)

<command-1981093759856396> in <module>

88

89

---> 90 main()

<command-1981093759856396> in main()

5 def main():

6 # Create streaming environment

----> 7 env = StreamExecutionEnvironment.get_execution_environment()

8

9 settings = EnvironmentSettings.new_instance()\

/databricks/python/lib/python3.8/site-packages/pyflink/datastream/stream_execution_environment.py in get_execution_environment()

801 """

802 gateway = get_gateway()

--> 803 j_stream_exection_environment = gateway.jvm.org.apache.flink.streaming.api.environment\

804 .StreamExecutionEnvironment.getExecutionEnvironment()

805 return StreamExecutionEnvironment(j_stream_exection_environment)

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/py4j/java_gateway.py in __getattr__(self, name)

1690 return UserHelpAutoCompletion()

1691

-> 1692 answer = self._gateway_client.send_command(

1693 proto.REFLECTION_COMMAND_NAME +

1694 proto.REFL_GET_UNKNOWN_SUB_COMMAND_NAME + name + "\n" + self._id +

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/py4j/java_gateway.py in send_command(self, command, retry, binary)

1029 if `binary` is `True`.

1030 """

-> 1031 connection = self._get_connection()

1032 try:

[-werners-]

did you get Flink running on the Databricks cluster? Because that seems to be the issue here.

[witnessthee]

Thanks for reply. May I know how to run Flink on Databricks clusters?

[-werners- · Accepted solution]

I have no idea tbh, but it seems like a bad idea. Databricks = Spark.

If you want to use Flink, I would run a Flink cluster.

If you are ok with spark, you do not need the pyflink library. Databricks has a Kakfa connector:

https://docs.databricks.com/structured-streaming/kafka.html

[Kaniz]

Hi @Tam daad​, We haven’t heard from you since the last response from @Werner Stinckens​, and I was checking back to see if you have a resolution yet.

If you have any solution, please share it with the community as it can be helpful to others. Otherwise, we will respond with more details and try to help.

Also, Please don't forget to click on the "Select As Best" button whenever the information provided helps resolve your question.