Spark本地与hdfs性能答案

【问题标题】：Spark local vs hdfs permormanceSpark本地与hdfs性能
【发布时间】：2016-04-18 05:38:57
【问题描述】：

我在同一台机器上有一个 Spark 集群和一个 HDFS。我在每台机器的本地文件系统和 hdfs 分布式文件系统上复制了一个大约 3GB 的文本文件。

我有一个简单的字数统计 pyspark 程序。

如果我提交从本地文件系统读取文件的程序，它会持续大约 33 秒。如果我提交从 hdfs 读取文件的程序，它会持续大约 46 秒。

为什么？我期待完全相反的结果。

在 sgvd 的请求后添加：

16 个奴隶 1 个主人

Spark Standalone，无特定设置（复制因子 3）

版本 1.5.2

import sys
sys.path.insert(0, '/usr/local/spark/python/')
sys.path.insert(0, '/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip')
import os
os.environ['SPARK_HOME']='/usr/local/spark'
os.environ['JAVA_HOME']='/usr/local/java'
from pyspark import SparkContext
#conf = pyspark.SparkConf().set<conf settings>


if sys.argv[1] == 'local':
    print 'Esecuzine in modalita local file'
    sc = SparkContext('spark://192.168.2.11:7077','Test Local file')
    rdd = sc.textFile('/root/test2')
else:
    print 'Esecuzine in modalita hdfs'
    sc = SparkContext('spark://192.168.2.11:7077','Test HDFS file')
    rdd = sc.textFile('hdfs://192.168.2.11:9000/data/test2')


rdd1 = rdd.flatMap(lambda x: x.split(' ')).map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y)
topFive = rdd1.takeOrdered(5,key=lambda x: -x[1])
print topFive

【问题讨论】：

它可能取决于很多事情。你的集群有多大？你用什么集群管理器？有什么自定义设置吗？什么火花版本？你能展示你的代码吗？
我在问题的空间里回答。

标签： performance hadoop apache-spark

【解决方案1】：

这有点违反直觉，但由于复制因子为 3，并且您有 16 个节点，因此每个节点平均有 20% 的数据本地存储在 HDFS 中。那么平均大约 6 个工作节点应该足以读取整个文件而无需任何网络传输。

如果您记录运行时间与工作节点数量的关系，您应该注意到大约 6 点之后，从本地 FS 和 HDFS 读取将没有区别。

上述计算可以使用变量来完成，例如x=number of worker nodes，y= replication factor，但是您可以很容易地看到，由于从本地 FS 读取会强制该文件位于所有节点上，因此您最终会得到 x=y，并且在使用 floor(x/y) 节点之后不会有任何区别。这正是您所观察到的，乍一看似乎违反直觉。你会在生产中使用 100% 的复制因子吗？

【讨论】：

改变代表因子但不改变工人数量不会改变时间。使用 6 Worker repfactor 3 和 6 datanode 时间增加到 1 分 30 秒。
你是怎么配置的？您是否重新启动了集群？你在描述中说你有 16 个奴隶。
Rep fact 更改尝试：我已将文件的 rep fact 从 2 更改为 16。程序提交给 16 个从站。尝试的节点数：我已经重新配置了整个集群（spark 和 hadoop），只有 6 个节点。

【解决方案2】：

Executor、Driver 和 RDD 的特定参数是什么（关于 Spilling 和存储级别）？

来自 Spark documentation

性能影响

The Shuffle is an expensive operation since it involves disk I/O, data serialization, and network I/O. 为了组织 shuffle 的数据，Spark 生成任务集 - map 任务来组织数据，以及一组 reduce 任务来聚合它。此命名法来自 MapReduce，与 Spark 的 map 和 reduce 操作没有直接关系。

某些 shuffle 操作会消耗大量堆内存，因为它们使用内存中的数据结构在传输记录之前或之后组织记录。 Specifically, reduceByKey and aggregateByKey create these structures on the map side, and 'ByKey operations generate these on the reduce side. When data does not fit in memory Spark will spill these tables to disk, incurring the additional overhead of disk I/O and increased garbage collection.

我对 Spark Job 的 memory/CPU core 限制与 Map & Reduce 任务的 memory/CPU core 限制感兴趣。

从 Hadoop 进行基准测试的关键参数：

yarn.nodemanager.resource.cpu-vcores
mapreduce.map.cpu.vcores
mapreduce.reduce.cpu.vcores
mapreduce.map.memory.mb
mapreduce.reduce.memory.mb
mapreduce.reduce.shuffle.memory.limit.percent

针对 Hadoop 对 SPARK 参数进行基准测试以实现等效性的关键参数。

spark.driver.memory
spark.driver.cores
spark.executor.memory
spark.executor.cores
spark.memory.fraction

这些只是一些关键参数。看看SPARK和Map Reduce的详细设置

如果没有正确的参数集，我们就无法比较两种不同技术的作业性能。

【讨论】：

【解决方案3】：

这是因为数据是如何分布的，单个文档不是一个好的选择，有几个更好的选择，比如parquet，如果你这样做你会注意到性能会明显提高，这是因为文件的分区方式允许您的Apache Spark 集群并行读取这些部分，从而提高性能。

【讨论】：