Programmiererforum

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

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)

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]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
[/code]
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]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)
[/code]
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!