[Python] Can not save spark 4.0 sql dataframe into redis database (Connection reset by peer)

I try to make some python codes of saving spark sql dataframe into redis database. First, my development environments are

OS : Windows 11
Spark : Spark 4.0.0
python : Anaconda3-2024.02-1-Windows-x86_64
IDE : Visual Studio Code


And These are simple pyspark codes.

from pyspark.sql import SparkSession
from pyspark.sql.functions import to_json, struct

import redis

def save_to_redis_partition(rows):
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=False)

for row in rows:
key = f"user:{row['id']}"
value = row['json']
r.set(key, value)

r.close()

def main():
spark = SparkSession.builder \
.appName("PySpark Jedis Redis Demo") \
.master('local[*]')\
.getOrCreate()

data = [(1, "Alice", 30), (2, "Bob", 25), (3, "Charlie", 35)]
df = spark.createDataFrame(data, schema=["id","name","age"])

df_json = df.withColumn("json", to_json(struct("id","name","age")))

df_json.select("id","json").rdd.map(lambda x: x.asDict()).foreachPartition(save_to_redis_partition)

spark.stop()

if __name__ == "__main__":
main()


But Errors are thrown when inserting spark dataframe into redis.

Caused by: java.io.IOException: Connection reset by peer
at java.base/sun.nio.ch.SocketDispatcher.write0(Native Method)
at java.base/sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:54)
at java.base/sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:132)
at java.base/sun.nio.ch.IOUtil.write(IOUtil.java:76)
at java.base/sun.nio.ch.IOUtil.write(IOUtil.java:53)
at java.base/sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:532)
at org.apache.spark.api.python.BasePythonRunner$ReaderInputStream.writeAdditionalInputToPythonWorker(PythonRunner.scala:855)
at org.apache.spark.api.python.BasePythonRunner$ReaderInputStream.read(PythonRunner.scala:767)
at java.base/java.io.BufferedInputStream.fill(BufferedInputStream.java:244)
at java.base/java.io.BufferedInputStream.read(BufferedInputStream.java:263)
at java.base/java.io.DataInputStream.readInt(DataInputStream.java:393)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:933)


The below messages are written In log file.

"message":"Executor is trying to kill task 7.0 in stage 0.0 (TID 7), interruptThread: false, reason: Stage cancelled: Job aborted due to stage failure: Task 2 in stage 0.0 failed 1 times, most recent failure: Lost task 2.0 in stage 0.0 (TID 2) (DESKTOP-D8UU19A.localdomain executor driver): org.apache.spark.SparkException: Python worker exited unexpectedly (crashed)


However, above codes are executed successfully on spark 4.0 installed on Ubuntu 22.04. So I have no idea why these errors occurs only on windows 11. Do I miss any configuration of spark 4.0 on windows 11? Any reply will be grateful.

Continue reading...