使用Google CloudDataproc时是否仍需要微调spark配置参数？

Question

详细说明：

• 通常，在编写spark作业时，需要为不同的spark配置指定特定值，以便以最佳方式使用群集资源。我们可以在初始化SparkSession时以编程方式执行此操作： SparkSession.builder .appName（SPARK_APP_NAME）.config（“spark.executor.memory”，“1G”）
• 我想知道的是：使用Cloud Dataproc时我们还需要这样做吗？实际上，在创建Dataproc集群时，会初始化名为cluster.properies的属性文件，并包含spark\:spark.executor.memory=2688m等值。所以，我想知道Dataproc是否会自动以最佳方式填充这些值w.r.t.群集资源，在这种情况下，我们不必手动/编程调整这些火花配置？

Answer 1

Dataproc确实提供了基于机器类型（甚至是自定义机器类型）和集群形状的智能默认值，这些默认值旨在成为最佳“一刀切”设置，以平衡每个JVM的更多线程的效率与共享资源池的限制每个JVM;粗略地说，机器被分割出来以适应每台机器的2个执行器，并且每个执行器被给予半个机器的线程（因此你希望2个执行器能够在n1-standard-8上并行运行4个任务，例如）。

请记住，YARN对于incorrectly report vcores for multi-threaded Spark executors是众所周知的，所以在Dataproc上运行一个大的Spark工作时，你可能只看到两个YARN“vcores”，但你可以通过查看Spark AppMaster页面来验证所有核心是否确实被使用，运行ps工作者，或在Dataproc云控制台页面上查看CPU使用情况。

但是，这些类型的设置从不普遍100％“最佳”，Dataproc尚未根据您运行的实际工作负载或历史工作负载自动预测设置。因此，纯粹基于群集形状的任何设置对于在该群集上运行的所有工作负载而言都不是100％最佳。

简而言之，在Dataproc上你不应该担心在大多数情况下显式优化，除非你试图真正挤出每一盎司的效率，但同时你可以随时用你自己的方式覆盖Dataproc的设置如果需要，可以在创建群集或作业提交时使用属性。需要考虑以下几点：

• 如果你有更多的cpu-heavy vs内存工作负载，考虑使用“highcpu”机器类型，Dataproc将自动让每个执行器为每个CPU分配更少的内存。
• 如果您的内存负载更大，请考虑使用highmem类型
• 如果您的输入不可分割和/或高度压缩（如.csv.gz文件），您可能更容易遇到内存问题，因为通常的并行计算不会知道输入数据是否会爆炸大于预期。在这些情况下，您可能需要覆盖执行程序内存以使其更大
• 如果您正在使用子进程或本机库，例如从工作任务调用ffmpeg，那么任务将消耗YARN / Spark知识之外的物理内存;在这些情况下，您可能还需要通过减少每个执行程序的内核或启动执行程序内存开销来调整内存限制。
• 如果你有一些非常IO绑定的东西或其他异步函数阻塞（比如从你的任务中调用一个缓慢的外部Web端点），那么你可能会从每个执行程序的核心启动中获益;然后Spark运行的任务比CPU多，但如果任务只是等待IO，这将有利于提高效率。

Answer 2

答案是肯定的。它取决于您的spark应用程序的行为，您运行的vm数以及您使用的vm类型。以下是我的示例调整参数。

default_parallelism=512

PROPERTIES="\
spark:spark.executor.cores=2,\
spark:spark.executor.memory=8g,\
spark:spark.executor.memoryOverhead=2g,\
spark:spark.driver.memory=6g,\
spark:spark.driver.maxResultSize=6g,\
spark:spark.kryoserializer.buffer=128m,\
spark:spark.kryoserializer.buffer.max=1024m,\
spark:spark.serializer=org.apache.spark.serializer.KryoSerializer,\
spark:spark.default.parallelism=${default_parallelism},\
spark:spark.rdd.compress=true,\
spark:spark.network.timeout=3600s,\
spark:spark.rpc.message.maxSize=256,\
spark:spark.io.compression.codec=snappy,\
spark:spark.shuffle.service.enabled=true,\
spark:spark.sql.shuffle.partitions=256,\
spark:spark.sql.files.ignoreCorruptFiles=true,\
yarn:yarn.nodemanager.resource.cpu-vcores=8,\
yarn:yarn.scheduler.minimum-allocation-vcores=2,\
yarn:yarn.scheduler.maximum-allocation-vcores=4,\
yarn:yarn.nodemanager.vmem-check-enabled=false,\
capacity-scheduler:yarn.scheduler.capacity.resource-calculator=org.apache.hadoop.yarn.util.resource.DominantResourceCalculator
  "

gcloud dataproc clusters create ${GCS_CLUSTER} \
       --scopes cloud-platform \
       --image pyspark-with-conda-v2-365 \
       --bucket  spark-data \
       --zone  asia-east1-b  \
       --master-boot-disk-size  500GB \
       --master-machine-type n1-highmem-2 \
       --num-masters  1 \ 
        --num-workers  2 \
       --worker-machine-type n1-standard-8 \
       --num-preemptible-workers 2 \
       --preemptible-worker-boot-disk-size 500GB \
       --properties ${PROPERTIES}