Warum wird nur einer von Spark-Jobs mit nur einem Executor ausgeführt?

Question

Mein Spark Cluster hat 1 Master und 2 Arbeiter. Die Anwendung liest CSV-Dateien von s3 in DataFrames, registriert sie als temporäre Tabellen und verwendet sqlContext, um SQL-Abfragen zum Erstellen neuer Datenrahmen auszuführen. Dann werden diese DFs in MySql DB gespeichert. Diese Jobs laufen alle auf mehreren Knoten.

Aber wenn ich diese Tabellen von DB zurück zu DataFrames, registrieren Sie sie als temporäre Tabellen und führen Sie sqlContext Abfrage, wird die gesamte Verarbeitung von nur einem Knoten erledigt. Was könnte das verursachen? Hier

ist ein Beispiel für meine Code:

DataFrame a = sqlContext.read().format("com.databricks.spark.csv").options(options) 
       .load("s3://s3bucket/a/part*"); 
DataFrame b = sqlContext.read().format("com.databricks.spark.csv").options(options) 
       .load("s3://s3bucket/b/part*"); 

a.registerTempTable("a"); 
b.registerTempTable("b"); 

DataFrame c = sqlContext.sql("SELECT a.name, b.name from a join b on a.id = b.a_id"); 

c.write().mode(SaveMode.Append).jdbc(MYSQL_CONNECTION_URL, "c", prop); 

// other jobs are similar 

Map<String, String> dOptions = new HashMap<String, String>(); 
dOptions.put("driver", MYSQL_DRIVER); 
dOptions.put("url", MYSQL_CONNECTION_URL); 

dOptions.put("dbtable", "(select * from c) AS c"); 
rC= sqlContext.read().format("jdbc").options(dOptions).load(); 
rC.cache(); 

dOptions.put("dbtable", "(select * from d) AS d"); 
rD= sqlContext.read().format("jdbc").options(dOptions).load(); 
rD.cache(); 

dOptions.put("dbtable", "(select * from f) AS f"); 
rF= sqlContext.read().format("jdbc").options(dOptions).load(); 
rF.cache(); 

rC.registerTempTable("rC"); 
rD.registerTempTable("rD"); 
rF.registerTempTable("rF"); 

DataFrame result = sqlContext.sql("SELECT rC.name, rD.name, rF.date from rC join rD on rC.name = rD.name join rF on rC.date = rF.date"); 

result.write().mode(SaveMode.Append).jdbc(MYSQL_CONNECTION_URL, "result_table", prop); 

Answer 1

Könnten Sie mit uns teilen Sie Ihre SparkConf() Objekt?

Ein SparkConf() - Objekt enthält die Konfiguration für eine Spark-Anwendung. Es werden verschiedene Parameter als Funkenschlüsselwertpaare wie einzustellen:

-Das Master

-Anzahl der Vollstrecker

-Anzahl Zieher Kerne

-heap Speicher zugewiesen

-other ..