gleichzeitiger append auf HDFS-Datei in Funken

Question

ich ex erhalten, die Datei beschäftigt hdfs_non_map_reduce

ist Append_File gescheitert Ich nehme Aufzeichnungen von kafka durch Funken und steckt es in cassandra und hdfs stream.map(somefunc).saveToCassandra

stream.map(somefunc).foreachRDD(rdd => 
fs.append.write(rdd.collect.mkstring.getBytes) 
fs.close) 

Replikationsfaktor in hdfs ist 1, ich benutze einen Knoten-Cluster Funken Standalone-Cluster mit 2 Arbeiter

Ich möchte nicht rdd.toDF.save("append"), weil es eine Menge von Dateien macht. Irgendwelche Ideen. Oder möglicherweise hdfs hat Methode zu überprüfen, wenn die Datei beschäftigt ist, eine andere Aufgabe?

Answer 1

ich will nicht rdd.toDF.save („anhängen“), weil es eine Menge von Dateien

Verwenden rdd.repartition(1).toDF.save("append") die Anzahl der Ausgabedateien 1

Answer 2

Diese zu reduzieren macht, ist auch nicht gut für mich, es macht Datei für jede rdd, aber ich möchte eine große Datei für und Stunde oder Tag

So, jetzt benutze ich versuchen fangen schließlich Schema auf meinem Cluster

try { 
fs.append.write(rdd.collect.mkstring.getBytes) 
} 
catch { 
case ex: IOException => fs.wait(1000) 
} 
finally { 
fs.close 
} 

aber ich denke, dass ich Ausnahmen habe, aber es funktioniert einwandfrei, ich schreibe 100k msg zu kafka und feile auf hdfs habe auch so, und so kontrolliere ich das, aber ich dachte, dass auf diese Weise, wenn ex, msgs nicht schreiben, und fs.close