spark (java) - Zu viele geöffnete Dateien

Question

Ich versuche, einen Stapeljob in spark2 auszuführen, der eine riesige Liste als Eingabe nimmt und auf der Liste iteriert, um die Verarbeitung durchzuführen. Das Programm führt für rund 8 000 Datensätze der Liste in Ordnung und dann bricht geben Ausnahme:

WARN Lost task 0.0 in stage 421079.0 (TID 996338, acusnldlenhww4.cloudapp.net, executor 1): java.io.FileNotFoundException: /data/1/hadoop/yarn/local/usercache/A2159537-MSP01/appcache/application_1497532405817_0072/blockmgr-73dc563c-8ea5-4f2d-adfe-6c60cf3e3968/0d/shuffle_145960_0_0.index.cfb6d5ea-8c7b-41a1-acc3-2c840e7f8998 (Too many open files) 
     at java.io.FileOutputStream.open0(Native Method) 
     at java.io.FileOutputStream.open(FileOutputStream.java:270) 
     at java.io.FileOutputStream.<init>(FileOutputStream.java:213) 
     at java.io.FileOutputStream.<init>(FileOutputStream.java:162) 
     at org.apache.spark.shuffle.IndexShuffleBlockResolver.writeIndexFileAndCommit(IndexShuffleBlockResolver.scala:144) 
     at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:128) 
     at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96) 
     at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53) 
     at org.apache.spark.scheduler.Task.run(Task.scala:99) 
     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322) 
     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
     at java.lang.Thread.run(Thread.java:745) 
    (org.apache.spark.scheduler.TaskSetManager) 

Neo4j Datenbank als Eingabe verwendet wird. Ich lese 300k Knoten von neo4j als Eingabe und mache eine for-Schleife auf der Eingabe rdd.

Versuchte Einstellung spark.shuffle.consolidateFiles zu true in SparkConf. Aber das hat nicht funktioniert.

Answer 1

Erhöhen Sie die ulimit wenn möglich - um dies zu überwinden.

Verringern Sie die Anzahl der Reduzierungen oder Kerne, die von jedem Knoten verwendet werden. Aber es hat Auswirkungen auf die Leistung Ihres Jobs.

Im Allgemeinen, wenn der Cluster:

assigned cores = `n`; 

und führen Sie einen Job mit:

reducers = `k` 

dann n * k Dateien parallel öffnet Funken und losschreiben.

Der Standardwert ulimit ist: 1024, der für Anwendungen im großen Maßstab zu niedrig ist.

Verwenden Sie ulimit -a, um die aktuelle maximale Anzahl geöffneter Dateien anzuzeigen.

Wir können die Anzahl der geöffneten Dateien vorübergehend ändern; durch Aktualisieren der Systemkonfigurationsdateien.

Sehen Sie diese Dateien für das gleiche:

/etc/sysctl.conf 
/etc/security/limits.conf