java - 为什么只有一个 Spark 作业仅使用一个执行器运行？

Question

我的 Spark 集群有 1 个主节点和 2 个工作节点。应用程序将 csv 文件从 s3 读取到 DataFrame，将它们注册为临时表，并使用 sqlContext 运行 sql 查询来创建新的 DataFrame。然后这些DF被存储到MySql DB中。这些作业都在多个节点上运行。

但是当我将这些表从数据库读回到 DataFrames、将它们注册为临时表并运行 sqlContext 查询时，所有处理仅由一个节点完成。可能是什么原因造成的？

这是我的代码示例:

 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

您可以与我们分享您的 SparkConf() 对象吗？

SparkConf() 对象包含 Spark 应用程序的配置。它用于将各种 Spark 参数设置为键值对，例如:

-大师

-执行者数量

-执行器核心数

-堆内存分配

-其他..