为什么行计数作业在 spark shell 中比 mapreduce 作业运行得慢

Question

我做了一个测试来比较 spark 和 mapreduce 之间的性能。我有三个节点集群，每个节点有 128GB 内存。

我运行一个作业来计算一个 10GB 文件中有多少行。

我使用默认配置的 hadoop 使用 mapreduce 运行行计数作业。我只需要大约 23 秒。

当我在每个节点 8GB 内存的 spark shell 中运行行计数作业时，我花了超过 6 分钟，这真的让我吃惊。

这里是启动spark-shell的命令和spark job的代码。

spark-shell --master  spark://10.8.12.16:7077 --executor-memory 8G
val s= sc.textFile("hdfs://ns/alluxio/linecount/10G.txt")
s.count()

我的 spark 配置文件来了：

spark-env.sh

export JAVA_HOME=/home/appadmin/jdk1.8.0_77
export SPARK_HOME=/home/appadmin/spark-2.0.0-bin-without-hadoop
export HADOOP_HOME=/home/appadmin/hadoop-2.7.2
export SPARK_DIST_CLASSPATH=$(/home/appadmin/hadoop-2.7.2/bin/hadoop classpath)
export YARN_CONF_DIR=$HADOOP_HOME/etc/hadoop
export SPARK_LIBARY_PATH=.:$JAVA_HOME/lib:$JAVA_HOME/jre/lib:$HADOOP_HOME/lib/native
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
SPARK_MASTER_HOST=10.8.12.16
SPARK_MASTER_WEBUI_PORT=28686
SPARK_LOCAL_DIRS=/home/appadmin/spark-2.0.0-bin-without-hadoop/sparkdata/local
SPARK_WORKER_MEMORY=10g


SPARK_WORKER_DIR=/home/appadmin/spark-2.0.0-bin-without-hadoop/sparkdata/work
SPARK_LOG_DIR=/home/appadmin/spark-2.0.0-bin-without-hadoop/logs

spark-default.conf

spark.driver.memory              5g
spark.eventLog.dir      hdfs://10.8.12.16:9000/spark-event-log

Answer 1

您可以传递分区数，即 defaultMinPartitions 调整分区数

喜欢这个

sc.textFile(file, numPartitions)
  .count()  

你也可以在加载后试试repartition看看效果。 另外，看看how-to-tune-your-apache-spark-jobs

您可以通过打印进一步调试和调整设置

sc.getConf.getAll.mkString("\n")

还可以像下面的示例 sn-p 一样获取执行器的数量。

/** Method that just returns the current active/registered executors
        * excluding the driver.
        * @param sc The spark context to retrieve registered executors.
        * @return a list of executors each in the form of host:port.
        */
       def currentActiveExecutors(sc: SparkContext): Seq[String] = {
         val allExecutors = sc.getExecutorMemoryStatus.map(_._1)
         val driverHost: String = sc.getConf.get("spark.driver.host")
         allExecutors.filter(! _.split(":")(0).equals(driverHost)).toList
       }

sc.getConf.getInt("spark.executor.instances", 1)

getExecutorStorageStatus 和getExecutorMemoryStatus 都返回包括驱动程序在内的执行器数量。

Answer 2

默认情况下，Spark 使用一个执行器运行。我会将我的设置更改为：

在每台机器上留出 8GB 空间用于操作系统和其他进程开销。那将为您留下120GB。考虑到如果你有超过 32GB 的垃圾收集器开始降级，我将在每台机器上有 4 个执行程序，每个执行程序 30GB。

所以，我会设置：

• spark.executor.instances = 12
• spark.executor.cores =（您的每台机器的核心数 - 1）/4（为操作系统保留 1）
• spark.executor.memory = 30g

然后再次运行您的应用程序。