我在下面的Github存储库中附加了一个链接，其中包含有关Spark和Docker入门的基本教程。 我从DockerHub的Getty镜像获得了Docker镜像。 分叉存储库后，运行scrip build_cluster.sh。

更具体地说，该脚本运行以下docker命令。

docker-compose up —-scale worker=$NUM_OF_WORKERS

在其中，docker-compose.yml引用了两个重要属性，即端口和暴露：

· 提及的端口将在docker-compose启动的不同服务之间共享。 端口将向主机公开一个随机端口或给定端口。 激活容器以侦听docker外部环境（可以是同一主机或其他计算机）以及docker内部的可访问世界中的指定端口。

· 基本上，"公开"端口无需公开就可以"公开"端口，而只有链接的服务才能访问它们。 只能指定内部端口。 激活容器以仅侦听docker内部的世界中的特定端口，而不侦听docker外部的世界中的特定端口。

Spark架构

Spark Master是一个用于协调从站资源分配的应用程序。 大师不执行任何计算。 师父只是一个资源经理。 Spark worker是工作程序节点上的应用程序，用于协调给定工作程序节点上的资源。 最后，Spark executor是由Spark Worker创建的应用程序，该应用程序在驱动程序的worker节点上执行任务。 在执行层次结构的顶部是Spark作业。 在Spark应用程序内部调用动作会触发Spark作业的启动以完成该作业。 执行计划包括将工作的转换组装到各个阶段。 阶段对应于全部执行相同代码的任务的集合，每个任务都在数据的不同子集上。 每个阶段都包含一系列转换序列，这些转换序列可以在不对完整数据进行混洗的情况下完成。

在每个阶段边界，父阶段的任务将数据写入磁盘，然后在子阶段的任务通过网络获取数据。 由于它们会占用大量磁盘和网络I / O，因此阶段边界可能很昂贵，应尽可能避免。 正如缩减器的数量是调整MapReduce作业的重要参数一样，调整阶段边界上的分区数量通常会影响或破坏应用程序的性能。 选择操作员安排时的主要目标是减少改组的次数和改组的数据量，因为改组很昂贵。 但是，偶尔减少随机播放次数的规则是有例外的。 但是，在特殊情况下，增加并行度会增加并行度，从而对性能有利。

Scala示例

Scala是一种静态类型的语言，它使我们能够发现编译时错误。 Spark是用Scala编写的，因为它是静态类型的，并且可以以已知的方式编译到JVM，因此速度非常快。 Scala通常比Python快10倍以上。 此外，Scala是基于JVM的Hadoop本机。 Hadoop很重要，因为Spark是在Hadoop的文件系统HDFS之上构建的。 Scala通过Java中的本机Hadoop API与Hadoop交互。 因此，在Scala中编写本机Hadoop应用程序非常容易。 与Python相比，Scala的高级功能可能使学习起来更加复杂。 但是在框架，库，隐式，宏等方面，Scala总是更强大。而且，由于其功能性质，Scala在MapReduce框架中运行良好。

#1. Calculate the Value of Pi

bin/run-example SparkPi 10

#2. Linear regression with elastic-net (mixing L1/L2) regularization

bin/run-example ml.LinearRegressionExample --regParam 0.15 --elasticNetParam 1.0 data/mllib/sample_linear_regression_data.txt

Python示例

Python是动态类型的，这会降低速度，并使其在每次更改代码时都容易出错。 编译语言比解释语言要快。 Spark库必须调用，这需要大量的代码处理，因此会降低性能。 Python与Hadoop服务的交互非常差，因此开发人员必须使用第三方库（如hadoopy）。 但是Python具有简单的语法和良好的标准库，因此对于简单的直观逻辑更可取，而Scala对于复杂的工作流程则更有用。 特别是对于复杂的机器学习应用程序（如NLP，图形和可视化），Python是首选，因为Scala并没有太多工具。

1. Calculate the Pagerank for a bunch of web page urls

bin/spark-submit examples/src/main/python/pagerank.py data/mllib/pagerank_data.txt 10

2. Demonstrate K-means clustering and calculate euclidian distance (requires numpy)

bin/spark-submit examples/src/main/python/ml/kmeans_example.py

扩大我们的Worker

最初，我的所有工作人员都会自动被分配到docker-compose文件中提到的端口80，并且由于该端口已被占用，因此容器出现故障。 因此，我要做的就是设置一个Traefik反向代理容器（它是一个负载平衡器），并将我的工作容器端口暴露给将处理扩展问题的负载平衡器。

reverse-proxy: image: traefik command: - "--api.insecure=true" - "--providers.docker=true" - "--providers.docker.exposedbydefault=false" - "--entrypoints.web.address=:80" ports: - "80:80" # The HTTP port - "8080:8080" # The Web UI (enabled by --api)

调整我们的Spark作业

Spark（和YARN）考虑的两个主要资源是CPU和内存。 应用程序中的每个Spark执行程序都具有相同的固定核心数和相同的固定堆大小。 cores属性控制执行程序可以运行的并发任务数。 内存属性会影响Spark可以缓存的数据量，以及用于分组，聚合和联接的混洗数据结构的最大大小。 运行内存过多的执行程序通常会导致过多的垃圾回收延迟。

这是docker-compose.yml文件中工作容器的设置。

environment:

SPARK_WORKER_CORES: 2

SPARK_WORKER_MEMORY: 1g

SPARK_PUBLIC_DNS: localhost

在行业中，对于单个执行器来说，64GB是一个不错的上限。 而且HDFS客户端在处理大量并发线程时遇到了麻烦，因此每个执行程序最多可以执行5个任务，以实现完整的写入吞吐量。 Spark是一个并行处理引擎，但是其计算出最佳并行度的能力有限。 每个Spark阶段都有许多任务，每个任务都按顺序处理数据。 在调整Spark作业时，此数字可能是确定性能的唯一最重要的参数。 此数字由Spark将RDD分为多个阶段的方式确定。

数据以记录的形式流经Spark。 一条记录具有两种表示形式：反序列化的Java对象表示形式和序列化的二进制表示形式。 通常，Spark将反序列化表示形式用于内存中的记录，将序列化表示形式用于存储在磁盘上或通过网络传输的记录。 这两种表示形式的记录足迹对Spark性能有很大影响。 膨胀的反序列化对象将导致Spark更频繁地将数据溢出到磁盘上，并减少反序列化记录的数量。

SequenceFile和小文件问题

最后，只要您有能力决定如何在磁盘上存储数据，请使用可扩展的二进制格式，例如Avro，Parquet，Thrift或Protobuf。 选择这些格式之一并坚持下去。 明确地说，当谈论在Hadoop上使用Avro，Thrift或Protobuf时，它们的意思是每条记录都是存储在序列文件中的Avro / Thrift / Protobuf结构。 JSON并不值得，因为会浪费大量精力来不断地花费大量的CPU周期来解析文件。

当许多小文件导致引用大量小文件的namenode的内存开销时，就会出现小文件问题。 SequenceFile的概念是将每个小文件放入一个较大的单个文件中。 例如，假设有10,000个大小为100KB的文件，那么我们可以编写一个程序将它们放入如上的单个SequenceFile中，在这里您可以使用filename作为键，而content作为值。

写入顺序文件以容纳多个键值对，并且键是唯一的文件元数据，例如摄取文件名或文件名+时间戳，而值是摄取文件的内容。 现在，您有一个文件，其中包含许多已摄取文件作为可拆分键值对。 因此，如果将其加载到pig或Hive中，并按键分组，则每个文件内容将是其自己的记录。 序列文件是包含键值对的二进制文件。 它们可以在记录（键值对）或块级别进行压缩。 Java API通常用于写入和读取序列文件，并且由于它们是二进制文件，因此它们的读取/写入速度比文本格式的文件要快。

(本文翻译自Suraj Patil的文章《Building a Scalable Spark cluster with Docker Containers》，参考：https://towardsdatascience.com/building-a-scalable-spark-cluster-with-docker-containers-f921d860fa46)