PySpark：设置执行程序/内核和内存本地计算机

如何解决PySpark：设置执行程序/内核和内存本地计算机

因此，我查看了Pyspark，Jupyter上的一堆帖子，并设置了内存/核心/执行程序（以及相关的内存）。

但是我似乎被困住了-

问题1：我没有看到我的计算机使用内核或内存。为什么？我可以对准考者/核心/内存进行一些调整以优化读取文件的速度吗？ 问题2：还有什么办法可以让我看到进度条，其中显示了导入了多少文件（spark-monitor似乎没有这样做）。

我正在将33.5gb文件导入pyspark。

机器具有112 GB或RAM 8核心/ 16虚拟核心。

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Summaries") \
    .config("spark.some.config.option","some-value") \
    .getOrCreate()
conf = spark.sparkContext._conf.setAll([('spark.executor.memory','4g'),('spark.app.name','Spark Updated Conf'),('spark.driver.cores','4'),('spark.executor.cores','16'),('spark.driver.memory','90g')])
spark.sparkContext.stop()
spark = SparkSession.builder.config(conf=conf).getOrCreate()
df = spark.read.json("../Data/inasnelylargefile.json.gz")

我假设pyspark即使在读取文件时也在发挥其魔力（所以我应该看到大量的内核/内存使用）。但是我没有看到它。帮助！

更新：使用较小的zip文件（89 MB）进行了测试

Pyspark需要72秒 熊猫需要10.6秒 使用的代码：

start = time.time()
df = spark.read.json("../Data/small.json.gz")
end = time.time()
print(end - start)

start = time.time()
df = pa.read_json('../Data/small.json.gz',compression='gzip',lines = True)
end = time.time()
print(end - start)

解决方法

尽管您的问题的答案仅在于以下问题之一，但让我重写您的示例以解释正在发生的事情。

设置您的配置

首先，您不需要启动和停止上下文来设置配置。从spark 2.0开始，您可以创建spark会话，然后设置配置选项。

from pyspark.sql import SparkSession
spark = (SparkSession.builder.appName("yourAwesomeApp").getOrCreate())
spark.conf.set("spark.executor.memory","40g")
spark.conf.set("spark.executor.cores","2")

读取数据

Spark将懒惰地评估DAG。 您所测量的时间不是数据加载到数据帧中的时间，而是JSON文件的架构推断。模式推断非常昂贵，您应该尝试通过设置数据的模式来避免这种情况。 您会发现以下两者之间在性能上有很大差异：

df = spark.read.json("../data/a_very_large_json.json.gz")

和

from pyspark.sql.types import (
    StructType,StringType,StructField,)
json_schema = schema = StructType([
    StructField('data',StructType([
        StructField("field1",StringType(),nullable=False),StructField("field2",StructField("field3",nullable=True),StructField("field4",StructField("field5",LongType(),])),])
df = spark.read.json("../data/a_very_large_json.json.gz",schema=json_schema)

如果提供了架构，则该指令应几乎立即生效。 正如另一位用户已经提到的那样，要执行任务，您需要进行一项活动，例如表演，表演，收集，坚持等等。

df.show()

您可以在配置上设置执行程序实例和核心的数量，但是这些实例的实际使用还取决于您的输入数据和执行的转换/操作。根据您的描述，我假设您正在独立模式下工作，因此默认使用一个执行程序实例（使用所有内核），并且应将执行程序内存设置为使用可用的实例。据我所记得，当您在独立模式下工作时，spark.executor.instances被忽略，而实际的执行程序数则取决于可用的内核数和spark.executor.cores

与熊猫的比较

如果仅使用一个节点，将数据加载到数据帧中，则spark和pandas之间的比较是不公平的。 Spark总是会有更高的开销。当您的数据集不适合一台计算机的内存并且您有多个节点来执行计算工作时，火花将闪耀。如果您对大熊猫感到满意，我想您可能会对Databricks的koalas感兴趣。

推荐

我更喜欢在应用程序外部设置执行细节（例如，使用spark-submit参数）。在极少数情况下，为了提高性能，您需要将其中一些设置为代码，但是对于Spark的每个新版本，this is less frequent。如果能够实现这一目标，则您的应用程序将更加面向未来，并且易于扩展。

,

spark.sparkContext.stop()
spark = SparkSession.builder.config(conf=conf).getOrCreate()
df = spark.read.json("../Data/inasnelylargefile.json.gz")

添加此内容：

df.show() 
##OR
df.persist()

您所做的比较不是苹果与苹果，spark会进行惰性评估，这意味着如果您不对操作进行任何操作，它只会编译并为DAG做好准备。

在Spark中，有两个概念，

1. 转化：懒惰地评估
2. 操作：（例如collect（），take（），show（），persist（））立即进行评估。

在您的情况下，read（）只是一个转换，添加一个动作应该触发计算。

有关操作与转换的更多信息：https://training.databricks.com/visualapi.pdf

,

Spark读取速度比Pandas慢的原因是因为gz文件不可拆分，因此Spark必须通过单个任务读取整个文件。但是，当读取未压缩的文件或使用可拆分压缩格式压缩的文件（如bzip2）压缩时，Spark会并行部署x个任务（最多可达集群中可用核心的数量）以读取文件。尝试解压缩文件，然后再将其传递给Spark。