PySpark FileAlreadyExistsException: Ausgabeverzeichnis kann während saveAsTextFile nicht überschrieben werden ⇐ Python

[Guest]

Ich arbeite an einem PySpark-Skript, um eine einfache Wortzählung durchzuführen. Mein Skript läuft einwandfrei, aber beim Versuch, die Ergebnisse mit saveAsTextFile zu speichern, tritt ein Fehler auf (jetzt bin ich auf Ubuntu). Hier ist der Fehler, den ich erhalte:

Code: Select all

py4j.protocol.Py4JJavaError: An error occurred while calling o48.saveAsTextFile.
org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory file:/home/pyspark_python/wordcount/output_new already exists

Hier sind die Schritte, die ich bisher unternommen habe:
Überprüft, dass das Ausgabeverzeichnis keine Daten enthält (ls zeigt an, dass es leer ist).
Das Verzeichnis wurde mit rm -r und mkdir -p gelöscht und neu erstellt.
Es wurde sichergestellt, dass keine anderen Spark-Jobs ausgeführt werden (ps aux | grep spark).
Trotzdem , der Fehler bleibt bestehen, wenn ich es erneut ausführe das Skript.
Hier ist der Code, den ich verwende:

Code: Select all

from pyspark import SparkConf, SparkContext
import os

def main(input_file, output_dir):
# Configuration Spark
conf = SparkConf().setAppName("WordCountTask").setMaster("local[*]")
sc = SparkContext(conf=conf)

# Lecture du fichier d'entrée
text_file = sc.textFile(input_file)

# Comptage des mots
counts = (
text_file.flatMap(lambda line: line.split(" "))
.map(lambda word: (word, 1))
.reduceByKey(lambda a, b: a + b)
)

# Sauvegarde des résultats
if not os.path.exists(output_dir):
os.makedirs(output_dir)
counts.saveAsTextFile(output_dir)

print(f"Résultats sauvegardés dans le répertoire : {output_dir}")

if __name__ == "__main__":
# Définir les chemins d'entrée et de sortie
input_file = r"/home/othniel/pyspark_python/wordcount/input/loremipsum.txt"
output_dir = "/home/othniel/pyspark_python/wordcount/output_new"

# Exécution de la tâche WordCount
main(input_file, output_dir)

Wie kann ich diesen Fehler beheben und sicherstellen, dass PySpark erfolgreich in das Ausgabeverzeichnis schreibt? Muss ich in meinem Skript oder meiner Umgebung etwas Bestimmtes konfigurieren?
Vielen Dank für Ihre Hilfe!