皮皮网

1.Spark异常处理——Shuffle FetchFailedException
2.Spark 的源码 shuffle 流程以及寻址流程
3.Spark对shuffle阶段的优化以及调优
4.Spark Shuffle概念及shuffle机制
5.SparkShuffle及Spark SQL图解执行流程语法
6.Spark Shuffle的原理

Spark异常处理——Shuffle FetchFailedException

        在大规模数据处理中，这个错误比较常见。一般发生在有大量shuffle操作的时候，task不断的failed，然后又重执行，一直循环下去，直到application失败。

        SparkSQL shuffle报错样例

        RDD shuffle报错样例

        shuffle分为 shuffle write 和 shuffle read 两部分：

        解决办法主要从 shuffle的数据量和 处理shuffle数据的分区数两个角度入手。

        通过 spark.sql.shuffle.partitions 控制分区数，默认为，根据shuffle的量以及计算的复杂度提高这个值。

        通过 spark.default.parallelism 控制 shuffle read 与 reduce 处理的分区数，默认为运行任务的core的总数（mesos细粒度模式为8个，local模式为本地的core总数），官方建议设置成运行任务的core的2-3倍。

        通过 spark.executor.memory 适当提高executor的内存

        通过 spark.executor.cores 增加每个executor的cpu，这样不会减少task并行度

Spark 的 shuffle 流程以及寻址流程

       Spark的shuffle过程是分布式计算中关键的一环，它在数据重组时涉及复杂的源码细节。与MapReduce类似，源码shuffle连接Map和Reduce阶段，源码但Spark采用DAG调度，源码将宽依赖（shuffle）划分成不同的源码透视源码怎么使用Stage。在这个过程中，源码Map任务产生小文件并由MapOutPutTrackerMaster记录地址，源码Reduce任务在执行前通过MapOutputTracker获取这些地址，源码然后BlockManager通过ConnectionManager和BlockTransferService进行数据传输。源码

       具体寻址流程如下：map任务执行后，源码将文件地址封装在MapStatus中，源码汇报给Driver的源码MapOutputTrackerMaster。所有map任务完成后，源码Driver掌握了所有文件地址。源码reduce任务在开始时，通过MapOutputTracker获取这些地址，并通过BlockManager连接数据节点进行数据传输。默认情况下，数据会被分批拉取到Executor的shuffle聚合内存中，以避免内存溢出（OOM）。可通过减少数据量、增加shuffle聚合内存或Executor内存来避免这个问题。

       Spark的内存管理策略在1.6版本后有所改变，从静态管理转向统一管理，允许Storage和Execution共享内存，以实现更灵活的资源分配。关于shuffle的更多优化技巧，将在后续文章中深入探讨。

Spark对shuffle阶段的优化以及调优

       在大数据处理框架Apache Spark中，shuffle阶段是关键的性能瓶颈。传统MapReduce框架在shuffle阶段需要将Map任务的输出数据整理、合并，再传递给Reduce任务。Spark对此进行了优化，以提高效率。

       Map任务中，Spark使用内存缓冲区（默认MB）暂存输出数据。裂变模式源码当缓冲区接近满时，数据会溢写至磁盘，这称为“溢写”（Spill）。Spark有一个溢写阀值（spill.percent，默认0.8），当缓冲区使用率超过该阈值，Map任务会继续将数据写入剩余内存，同时执行排序和局部聚合（如果启用了Combiner）。所有溢写文件在Map任务结束时合并成一个文件。

       Reduce任务接收Map任务的输出文件，通过网络获取数据，然后在内存缓冲区中合并数据，如果内存缓冲区不足，数据同样可能溢写到磁盘。合并操作后，数据被写入最终的文件，这个过程称为“合并”（Merge）。

       优化后的Spark引入了SortShuffleManager，它有两种运行模式：普通模式和bypass模式。在普通模式下，数据先存储于内存数据结构中，根据shuffle算子类型（聚合或普通）选择不同的数据结构。当达到阈值时，数据被溢写到磁盘，并进行排序，分批写入文件，最后合并成一个文件。bypass模式下，每个下游任务对应一个磁盘文件，数据直接写入磁盘，无需内存缓冲，节省了排序步骤，提高了性能。

       调优方面，Spark提供了多个参数来优化shuffle阶段性能。如`spark.shuffle.file.buffer`、`spark.reducer.maxSizeInFlight`、寄存系统源码`spark.shuffle.io.maxRetries`、`spark.shuffle.io.retryWait`、`spark.shuffle.memoryFraction`、`spark.shuffle.manager`、`spark.shuffle.sort.bypassMergeThreshold`、`spark.shuffle.consolidateFiles`等。开发者需要根据实际情况调整这些参数，以获得最佳性能。

       简而言之，Spark通过改进shuffle机制，优化了数据传输过程，减少了文件数量，提高了读写效率，从而显著提升了整体处理速度。调优参数时，应结合实际工作负载、硬件资源和性能需求进行调整，以实现最佳性能表现。

Spark Shuffle概念及shuffle机制

       Spark Shuffle是连接Map与Reduce操作的关键步骤，它的性能直接影响到整个Spark程序的效率。在MapReduce中，shuffle涉及大量磁盘和网络I/O，而在Spark中，这个过程同样复杂，尤其是在DAG Scheduler的任务划分中，遇到宽依赖（shuffle）时，会划分一个新的Stage。

       Spark的shuffle过程涉及到几个核心组件，如MapOutPutTracker（主从架构的模块管理磁盘小文件地址）、BlockManager（主从架构的块管理，包括内存和磁盘管理）等。在Driver端和Executor端，BlockManager包含DiskStore、MemoryStore、ConnectionManager和BlockTransferService，它们负责数据的存储、管理与传输。mirai源码解析

       Spark的shuffle主要在reduceByKey等操作中发生，它将一个RDD中的数据按key聚合，即使key的值分布在不同分区和节点上。Shuffle Write阶段，map任务将相同key的值写入多个分区文件，而Shuffle Read阶段，reduce任务从所有map任务所在节点寻找相关分区文件进行聚合。

       Spark有HashShuffleManager和SortShuffleManager两种shuffle管理类型。HashShuffleManager在早期版本中采用普通（M * R）或优化（C * R）机制，而SortShuffleManager引入了排序和bypass机制。HashShuffle可能导致小文件过多和内存消耗问题，而SortShuffleManager则通过内存管理和排序优化，减少磁盘小文件数量。

       在执行流程中，map任务将结果写入缓冲，然后形成磁盘小文件，reduce task负责拉取并聚合这些小文件。然而，过多的小文件可能导致内存对象过多引发GC，甚至引发OOM。如果网络通信出现问题，可能导致shuffle过程中的数据丢失，此时由DAGScheduler负责重试Stage。

       HashShuffleManager的优化机制将磁盘小文件数量减少到C * R，而SortShuffleManager的普通和bypass机制分别产生2 * M和2 * M个磁盘小文件。SortShuffleManager的byPass机制只有在特定条件下才触发，以减少磁盘写入操作。

       总的来说，Spark的shuffle过程是一个复杂的操作，涉及数据的分布、聚合和传输，通过合理的shuffle策略和组件管理，以优化性能和避免潜在问题。

SparkShuffle及Spark SQL图解执行流程语法

       SparkShuffle是Apache Spark中的一个核心概念，主要涉及数据分片、聚合与分发的过程。在使用reduceByKey等操作时，问卷表单源码数据会被划分到不同的partition中，但每个key可能分布在不同的节点上。为了解决这一问题，Spark引入了Shuffle机制，主要分为两种类型：HashShuffleManager与SortShuffleManager。

       HashShuffleManager在Spark 1.2之前是默认选项，它通过分区器（默认是hashPartitioner）决定数据写入的磁盘小文件。在Shuffle Write阶段，每个map task将结果写入到不同的文件中。Shuffle Read阶段，reduce task从所有map task所在的机器上寻找属于自己的文件，确保了数据的聚合。然而，这种方法会产生大量的磁盘小文件，导致频繁的磁盘I/O操作、内存对象过多、频繁的垃圾回收（GC）以及网络通信故障，从而影响性能。

       SortShuffleManager在Spark 1.2引入，它改进了数据的处理流程。在Shuffle阶段，数据写入内存结构，当内存结构达到一定大小时（默认5M），内存结构会自动进行排序分区并溢写磁盘。这种方式在Shuffle阶段减少了磁盘小文件的数量，同时在Shuffle Read阶段通过解析索引文件来拉取数据，提高了数据读取的效率。

       Spark内存管理分为静态内存管理和统一内存管理。静态内存管理中内存大小在应用运行期间固定，统一内存管理则允许内存空间共享，提高了资源的利用率。Spark1.6版本默认采用统一内存管理，可通过配置参数spark.memory.useLegacyMode来切换。

       Shuffle优化涉及多个参数的调整。例如，`spark.shuffle.file.buffer`参数用于设置缓冲区大小，适当增加此值可以减少磁盘溢写次数。`spark.reducer.maxSizeInFlight`参数则影响数据拉取的次数，增加此值可以减少网络传输，提升性能。`spark.shuffle.io.maxRetries`参数控制重试次数，增加重试次数可以提高稳定性。

       Shark是一个基于Spark的SQL执行引擎，兼容Hive语法，性能显著优于MapReduce的Hive。Shark支持交互式查询应用服务，其设计架构对Hive的依赖性强，限制了其长期发展，但提供了与Spark其他组件更好的集成性。SparkSQL则是Spark平台的SQL接口，支持查询原生的RDD和执行Hive语句，提供了Scala中写SQL的能力。

       DataFrame作为Spark中的分布式数据容器，类似于传统数据库的二维表格，不仅存储数据，还包含数据结构信息（schema）。DataFrame支持嵌套数据类型，提供了一套更加用户友好的API，简化了数据处理的复杂性。通过注册为临时表，DataFrame的列默认按ASCII顺序显示。

       SparkSQL的数据源丰富，包括JSON、JDBC、Parquet、HDFS等。其底层架构包括解析、分析、优化、生成物理计划以及任务执行。谓词下推（predicate Pushdown）是优化策略之一，能够提前执行条件过滤，减少数据的处理量。

       创建DataFrame的方式多样，可以从JSON、非JSON格式的RDD、Parquet文件以及JDBC中的数据导入。DataFrame的转换与操作提供了灵活性和效率，支持通过反射方式转换非JSON格式的RDD，但不推荐使用。动态创建Schema是将非JSON格式的RDD转换成DataFrame的一种方法。读取Parquet文件和Hive中的数据均支持DataFrame的创建和数据的持久化存储。

       总之，SparkShuffle及Spark SQL通过高效的内存管理、优化的Shuffle机制以及灵活的数据源支持，为大数据处理提供了强大而高效的能力。通过合理配置参数和优化流程，能够显著提升Spark应用程序的性能。

Spark Shuffle的原理

       Spark Shuffle是数据处理中的关键环节，负责在Map和Reduce操作之间进行数据传输和排序。Hadoop Shuffle与Spark Shuffle有显著差异，Hadoop采用的是Push类型，流程包括map、spill、merge、shuffle和sort等步骤，而Spark提供两种主要的实现：Hash based Shuffle和Sort based Shuffle。

       Spark的Hash Shuffle在未优化时，数据可能会跨节点移动，但通过优化，如BypassMergeSortShuffleWriter，可以减少数据移动。相比之下，Sort Shuffle在Map端使用归并排序，输出索引文件指导Reduce端，但排序性能较低。Shuffle的性能受多种因素影响，如选择的Shuffle管理器（如hash、sort或tungsten-sort）、缓冲区大小（如shuffle.file.buffer和reducer.maxSizeInFlight）、重试策略（如maxRetries和retryWait）以及内存配置（如shuffle.memoryFraction）。

       Shuffle的选择和优化是提高Spark性能的关键，通过调整这些参数，可以平衡内存使用、磁盘I/O和网络通信，以实现更高效的中间结果传输和最终的聚合操作。

Spark Shuffle理解

       spark shuffle 演进的历史

        目前版本的shuffle, 都是使用排序相关的shuffle; 整体上spark shuffle分为shuffle read和shuffle write:

        大体上经过排序, 聚合, 归并(多个文件spill磁盘的情况), 最终, 每个task生成2种文件: 数据文件和索引文件.

        SortShuffleWriter是日常使用最频繁的shuffle过程; SortShuffleWriter主要使用 ExternalSorter 对数据进行排序, 合并, 聚合(combine). 最后产生数据文件和索引文件

            è¿™ä¸ªé—®é¢˜å°±æ˜¯ä¸Šè¿°æµç¨‹ä¸­, 第二点, MemoryManager怎么判断是否仍有内存空间留给内存中的shuffle write数据, 是否需要spill PartitionedAppendOnlyMap 和 PartitionedPairBuffer 的数据到磁盘? 这个问题的主要难处在于, spark内存中的数据都是有用数据, 往往无法通过GC自主控制内存, 所以如果spill时机检测的不及时, 即使产生GC可能仍会导致OOM问题. 但是如果每放入 PartitionedAppendOnlyMap 和 PartitionedPairBuffer

        中一条数据就检测内存占用情况, 会导致效率极其低下. Spark如何实现呢?

        我们说shuffle是可能会产生OOM的原因有2个:

        UnsafeShuffleWriter 是 SortShuffleWriter 的优化版本,Tungsten-sort优化点主要在三个方面:

        Spark 默认开启的是Sort Based Shuffle,想要打开Tungsten-sort ,请设置

        对应的实现类是：

SparkSpark Shuffle 原理

       探索 Spark Shuffle 的奥秘，今天继续深入学习 Spark Shuffle 原理。

       在 Spark 的执行流程中，划分 stage 时，根据是否涉及数据交换，可分为 ShuffleMapStage 和 ResultStage。深入理解这两种阶段有助于更好地掌握 Spark 的数据处理逻辑。

       接下来，我们将详细探讨 Spark Shuffle 的两种主要形式：HashShuffle 和 SortShuffle。

       HashShuffle 包括未优化和优化两个阶段。在未优化状态下，每个 Executor 线程仅执行一个 Task，而优化版本则通过复用缓冲区实现数据合并，此功能通过配置 spark.shuffle.consolidateFiles 开启。优化后的 HashShuffle 可大幅提高数据处理效率。

       SortShuffle 则采用不同的数据组织策略。普通 SortShuffle 中，数据首先以特定结构存储，然后通过溢写磁盘的方式进一步组织。最终，所有 Task 的数据聚集在一个文件内，并附带一份索引文件，标识每条数据的起始和结束 offset。

       在某些情况下，Spark 会启用 bypass SortShuffle，条件通常涉及特定的计算场景。此机制下，数据处理流程更为直接，减少了不必要的数据转移，从而提升性能。

       比较 HashShuffle 和 SortShuffle，前者在数据组织上更为灵活，而后者则在处理大容量数据时展现出色性能。此外，bypass SortShuffle 在特定条件下能提供更高效的处理方式。

       掌握 Spark Shuffle 的不同机制，对于优化 Spark 应用程序的性能至关重要。实践过程中，合理选择 Shuffle 方法能有效提升数据处理效率。

       如有兴趣了解更深入的 Spark 技术细节，欢迎关注。

Spark之Shuffle调优

       大多数Spark作业的性能关键在于shuffle环节，涉及大量的磁盘IO、序列化、网络数据传输。为了提升作业性能，shuffle调优至关重要。然而，性能优化整体而言，代码开发、资源参数配置和数据倾斜是关键因素，shuffle调优只占一小部分。因此，把握基本优化原则至关重要，避免本末倒置。下面将详细阐述shuffle原理、参数说明及调优建议。

       Spark运行分为两部分：驱动程序（SparkContext核心）和Worker节点上的Task。程序运行时，Driver与Executor间进行交互，包括任务分配、数据获取等，产生大量网络传输。Shuffle发生在下一个Stage向上游Stage请求数据时，即Stage间数据流动。

       ShuffleManager是负责shuffle过程执行、计算和处理的关键组件。在Spark 1.2版本后，从HashShuffleManager迭代为SortShuffleManager，显著减少了磁盘文件数量，提升性能。

       HashShuffleManager在shuffle write阶段，每个Task为下游Task创建大量磁盘文件，导致性能下降。SortShuffleManager则通过合并磁盘文件，每个Task拥有一个磁盘文件，减少磁盘IO操作，提升性能。

       优化HashShuffleManager的关键在于启用spark.shuffle.consolidateFiles参数，允许task复用磁盘文件，降低磁盘文件总数。

       SortShuffleManager运行机制分为普通和bypass两种。普通运行机制利用内存进行数据结构排序、批量写入磁盘，最后合并磁盘文件。bypass运行机制则直接将数据写入磁盘文件，简化过程，减少排序开销。

       在shuffle过程中，有多个关键参数需要优化，包括spark.shuffle.file.buffer、spark.reducer.maxSizeInFlight、spark.shuffle.io.maxRetries、spark.shuffle.io.retryWait、spark.shuffle.memoryFraction等。具体调优建议需根据实际作业性能测试和资源分配策略进行。

       Shuffle优化的目标在于减少磁盘IO操作，降低网络传输延迟，提升数据处理效率。合理配置上述参数，结合任务特性，能够显著提升Spark作业性能。