python - PySpark : Setting Executors/Cores and Memory Local Machine

Question

因此，我查看了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秒
Pandas 花了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)

Answer 1

尽管您问题的答案仅在于以下问题之一，但让我重写您的示例以解释正在发生的事情。
设定设定
首先，您不需要启动和停止上下文来设置配置。从spark 2.0开始，您可以创建spark session ，然后设置config选项。

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", StringType(), nullable=False),
        StructField("field3", StringType(), nullable=True),
        StructField("field4", StringType(), nullable=True),
        StructField("field5", LongType(), nullable=False),
    ])),
])
df = spark.read.json("../data/a_very_large_json.json.gz", schema=json_schema)

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

df.show()

您可以在配置上设置执行程序实例和核心的数量，但是这些实例的实际使用还取决于您的输入数据和执行的转换/操作。根据您的描述，我假设您正在独立模式下工作，因此默认具有一个执行程序实例(使用所有内核)，并且应将执行程序内存设置为使用可用的实例。据我所记得，当您在独立模式下工作时， spark.executor.instances会被忽略，而实际的执行程序数取决于可用的内核数和 spark.executor.cores与 Pandas 比较
如果仅使用一个节点，将数据加载到数据帧中，则spark和pandas之间的比较是不公平的。 Spark总是会有更高的开销。当您的数据集不适合一台计算机的内存并且您有多个节点来执行计算工作时， Spark 将闪耀。如果您对 Pandas 感到满意，我想您可能会对Databricks的 koalas感兴趣。
推荐
我更喜欢在应用程序外部设置执行细节(例如，使用spark-submit参数)。在极少数情况下，为了提高性能，您需要将其中一些设置为代码，但是使用Spark的每个新版本 this is less frequent。如果您能做到这一点，您的应用程序将更加面向 future ，并且易于扩展。