大规模的 数据分析 已经成为一个变革的工具 许多行业, 包括银行业欺诈检测在内的应用程序, 医疗保健临床研究, 以及制造业的预测性维护和质量控制. 然而, 处理如此大量的数据可能是一个挑战, 即使有现代计算硬件的力量. 现在有许多工具可用于解决这一挑战，其中最受欢迎的工具之一是 Apache火花这是一个开源的分析引擎，旨在加速处理非常大的数据集.

火花提供了一个强大的架构，能够处理大量的数据. 有几种火花优化技术可以简化流程和数据处理, 包括在内存中执行任务和在缓存中存储频繁访问的数据, 从而减少了检索期间的延迟. 火花 is also designed for scalability; data processing can be distributed across multiple computers, 提高可用的计算能力. 火花与许多项目相关:它支持各种编程语言(如java.g., Java, Scala, R, Python)，包括 各种库 (e.g., MLlib 对于机器学习， GraphX 用于处理图形，以及 火花流 用于处理流数据).

而火花的默认设置提供了一个很好的起点, 有几个调整可以提高它的性能，从而允许许多企业充分利用它的潜力. 在考虑火花中的优化技术时，有两个方面需要考虑:计算效率和优化节点之间的通信.

火花是如何工作的?

在详细讨论优化技术之前，了解一下火花如何处理数据是有帮助的. 火花的基本数据结构是弹性分布式数据集(RDD). 在考虑如何使用Apache火花时，理解rdd的工作原理是关键. RDD表示容错, 能够在一组计算机上并行处理的分布式数据集合. RDDs are immutable; their contents cannot be changed once they are created.

火花的快速处理速度是由rdd实现的. 虽然许多框架依赖于外部存储系统，如Hadoop分布式文件系统(HDFS)来重用和共享计算之间的数据, rdd支持内存计算. 在内存中执行处理和数据共享可以避免复制造成的大量开销, 序列化, 和磁盘读/写操作, 更不用说网络延迟了, 使用外部存储系统时. 火花通常被视为MapReduce的后继者 Hadoop来自Apache的早期框架. 虽然这两个系统共享类似的功能, 火花的内存处理允许它运行到 比MapReduce快100倍，它处理磁盘上的数据.

为了处理RDD中的数据，火花提供了一组丰富的 转换 和 行动. 转换使用以下操作从现有rdd中的数据生成新的rdd filter (), 加入(), or map (). 的 filter () 函数用满足给定条件的元素创建一个新的RDD，而 加入() 通过基于公共键组合两个现有RDD来创建新的RDD. map () 用于对数据集中的每个元素应用转换, 例如, 对RDD中的每条记录应用数学运算，例如计算百分比, 在新的RDD中输出结果. 一个动作, 另一方面, 不创建新的RDD, 但是返回对数据集进行计算的结果. 操作包括以下操作 count (), 第(), or 收集(). 的 count () action返回RDD中的元素个数，而 第() 只返回第一个元素. 收集() 简单地检索RDD中的所有元素.

转换进一步不同于操作，因为它们是惰性的. 转换的执行不是立即的. 而不是, 火花跟踪需要应用到基本RDD的转换, 而实际的计算只有在调用操作时才会触发.

Underst和ing RDDs 和 how they work can provide valuable insight into 火花 tuning 和 optimization; 然而, 尽管RDD是火花功能的基础, 对于许多应用程序来说，它可能不是最有效的数据结构.

选择正确的数据结构

而RDD是火花的基本数据结构, 它是一种低级API，需要更详细的语法，并且缺乏高级数据结构提供的优化. 随着数据框架的引入，火花转向了一个更加用户友好和优化的API，数据框架是建立在rdd之上的高级抽象. DataFrame中的数据被组织到命名列中, 将其结构更像关系数据库中的数据. DataFrame操作也受益于Catalyst、火花 SQL 优化的执行引擎，这可以提高计算效率，潜在地提高性能. 转换和操作可以像在rdd中那样在dataframe上运行.

因为它们的高级API和优化, DataFrames are typically easier to use 和 offer better performance; 然而, 由于它们的低级性质, rdd对于定义自定义操作仍然很有用, 以及调试复杂的数据处理任务. rdd提供了对分区和内存使用的更细粒度的控制. 处理原料时, 非组织性数据, 比如文本流, 二进制文件, 或者自定义格式, rdd可以更加灵活, 允许在没有预定义结构的情况下进行自定义解析和操作.

以下缓存最佳实践

缓存是一项重要的技术，可以显著提高计算效率. 可以缓存频繁访问的数据和中间计算, 或持续, 在允许更快检索的内存位置中. 火花提供了内置的缓存功能, 哪些对机器学习算法特别有益, 图形处理, 以及必须重复访问相同数据的任何其他应用程序. 没有缓存, 每次调用一个操作时，火花都会重新计算RDD或DataFrame及其所有依赖项.

下面的Python代码块使用 Py火花(火花的Python API)来缓存名为 df:

df.缓存()

重要的是要记住，缓存需要仔细规划, 因为它利用了火花工作节点的内存资源, 它们执行诸如执行计算和存储数据之类的任务. 如果数据集明显大于可用内存, 或者缓存了rdd或dataframe，而没有在后续步骤中重用它们, 潜在的溢出和其他内存管理问题可能会导致性能瓶颈.

优化火花的数据分区

火花的架构是围绕分区构建的, 将大量的数据分成小的数据, 更易于管理的单元称为分区. 分区使火花能够通过在多个节点上分布计算来并行处理大量数据, 每个处理总数据的一个子集.

而火花提供的默认分区策略通常基于可用CPU内核的数量, 它还提供了自定义分区的选项. 用户可以指定一个自定义分区函数，例如按某个键划分数据.

分区数

影响并行处理效率的最重要因素之一是分区的数量. 如果没有足够的分区，可用内存和资源可能没有得到充分利用. 另一方面, 由于任务调度和协调，分区太多可能导致性能开销增加. 分区的最佳数量通常设置为集群中可用内核总数的一个因子.

可以使用以下命令设置分区 重新分配() 和 合并(). 在这个例子中，DataFrame被重新划分为200个分区:

Df = Df.重新分区(200)	#重分区方法

Df = Df.合并(200)		# coalesce方法

的 重新分配() 方法增加或减少RDD或DataFrame中的分区数量，并在整个集群中执行数据的完全shuffle, 在处理和网络延迟方面，哪一种可能代价高昂. 的 合并() 方法减少RDD或DataFrame中的分区数量，而不像 重新分配(), 不执行一个完整的洗牌, 而是将相邻的分区组合起来以减少总数.

处理倾斜数据

在某些情况下, 某些分区可能比其他分区包含更多的数据, 导致一种被称为数据偏差的情况. 由于工作节点之间的工作负载分布不均匀，倾斜的数据可能导致并行处理效率低下. 为了解决火花中的倾斜数据，可以使用拆分或盐化等聪明的技术.

分裂

在某些情况下，倾斜分区可以被分割成多个分区. 如果某个数值范围导致数据偏斜, 这个范围通常可以分成更小的子范围. 例如, 如果很多学生在考试中得分在65%到75%之间, 考试成绩可以分为几个子范围, 比如65%到68%, 69% to 71%, 72%到75%.

如果某个特定的键值导致倾斜，则可以根据该键划分DataFrame. 在下面的示例代码中，数据中的倾斜是由大量具有 id “12345”的值.“ filter () 转换使用两次:一次选择所有带有 id 值“12345”，一次选择所有包含 id 不是“12345”.记录被放入两个新的dataframe: df_skew，它只包含具有 id 取值为“12345”，和 df_non_skew，它包含所有其他行. 数据处理可以在 df_skew 和 df_non_skew 分别，然后将得到的数据进行合并:

从pyspark.sql.函数导入r和

#根据歪斜键将数据框拆分为两个数据框.
Df_skew = df.Filter (df['id'] == 12345)	#包含id = 12345的所有行
Df_non_skew = df.过滤器(df(“id”) != 12345) #包含所有其他行

#将倾斜的DataFrame重新划分为更多分区.
Df_skew = Df_skew.重新分区(10)

#现在可以在两个dataframe上分别执行操作.
Df_result_skew = df_skew.groupBy (id).Count() #只是一个示例操作
Df_result_non_skew = df_non_skew.groupBy (id).count ()

#使用union()将操作的结果合并在一起.
Df_result = df_result_skew.联盟(df_result_non_skew)

盐

在分区之间更均匀地分布数据的另一种方法是向导致倾斜的键添加一个或多个“盐”. 盐的价值, 通常是一个随机数, 附加到原始密钥上, 盐渍键用于分区. 这迫使数据的分布更加均匀.

为了说明这个概念, 让我们想象一下，我们的数据被分成美国伊利诺伊州的三个城市:芝加哥的居民比附近的橡树公园或朗格罗夫城市的居民多得多, 导致数据失真.

为了更均匀地分布数据，使用Py火花，我们组合了列 城市 用一个随机生成的整数来创建一个新键，称为 salted_城市. " Chicago "变成了" Chicago ",”“Chicago2,和“芝加哥”,，每个新键代表较少数量的记录. 新键可用于操作或转换，例如 groupby () or count ():

在这个例子中，DataFrame 'df'有一个倾斜的列'城市'.
Skewed_column = '城市'

新建列salted_城市.
# 'salted_id'由原始'id'与后面添加0-10之间的随机整数组成
Df = Df.withColumn(“salted_城市”(df (skewed_column).Cast ("string") + (r和()*10).铸造(“int”).铸造(“字符串”)))

现在可以对“salted_城市”执行操作，而不是“城市”。.
假设我们正在做一个groupBy操作.
Df_grouped = df.groupby(“salted_城市”).count ()

#转化后，可以除去盐.
Df_grouped = Df_grouped.withColumn (original_城市, df_grouped(“salted_城市”).substr (0, len (df_grouped [' salted_城市 ']) 1))

广播

A 加入() 是基于一个或多个通用键组合两个数据集的通用操作吗. 通过匹配指定列中的值，可以将来自两个不同数据集的行合并为单个数据集. 由于需要跨多个节点进行数据变换，因此需要使用 加入() 就网络延迟而言，这是一个代价高昂的操作吗.

在一个小数据集与一个大数据集连接的场景中, 火花提供了一种称为广播的优化技术. 如果其中一个数据集足够小，可以装入每个工作节点的内存, 它可以发送到所有节点, 减少对代价高昂的洗牌操作的需求. 的 加入() 操作只是在每个节点上本地进行.

在下面的例子中，小的DataFrame df2 在所有工作节点上广播，而 加入() 操作与大数据框架 df1 在每个节点本地执行:

从pyspark.sql.函数导入广播
df1.加入(广播(df2),“id”)

df2 must be small enough to fit into the memory of each worker node; a DataFrame that is too large will cause out-of-memory errors.

过滤未使用的数据

在处理高维数据时，最小化计算开销至关重要. 任何不是绝对需要的行或列都应该被删除. 降低计算复杂度和内存使用的两个关键技术是早期过滤和列修剪:

早期的过滤: 应尽早在数据处理管道中应用过滤操作. 这减少了在后续转换中需要处理的行数, 减少总体计算负载和内存资源.

列修剪: 许多计算只涉及数据集中列的子集. 对于数据处理不需要的列应该删除. 列修剪可以显著减少需要处理和存储的数据量.

的示例 select () 用于修剪列的操作. 只有列 名字 和 年龄 加载到内存中. 该代码还演示了如何使用 filter () 操作使其仅包括值为的行 年龄 大于21:

Df = Df.选择(“名字”,“年龄”).filter(df['年龄'] > 21)

最小化Python用户定义函数的使用

Python用户定义函数(udf)是用Python编写的自定义函数，可以应用于rdd或dataframe. 和udf, users can define their own custom logic or computations; 然而, 有一些性能方面的考虑. 每次调用Python UDF, 数据需要在火花 JVM和Python解释器之间进行序列化和反序列化, 由于数据序列化，这会导致额外的开销, 进程切换, 数据复制. 这可能会显著影响数据处理管道的速度.

最有效的Py火花优化技术之一是尽可能使用Py火花的内置函数. Py火花附带了一个丰富的函数库，所有这些函数都经过了优化.

在复杂逻辑不能用内置函数实现的情况下, 使用向量化udf, 也称为P和as udf, 能帮助实现更好的表现吗. 向量化udf对整个列或数据数组进行操作，而不是对单个行进行操作. 这种批处理通常比逐行udf的性能更好.

考虑一个任务，其中列中的所有元素必须乘以2. 在下面的示例中，使用Python UDF执行此操作:

从pyspark.sql.函数导入udf
从pyspark.sql.types import IntegerType

def multiply_by_two (n):
   返回n * 2
multiply_by_two_udf = udf(multiply_by_two, IntegerType())
Df = Df.withColumn(“col1_doubled multiply_by_two_udf (df(“col1”)))

的 multiply_by_two () 函数是一个接受整数的Python UDF n 然后乘以2. 将此函数注册为UDF udf () 并应用到列上 col1 在数据框架内 df.

同样的乘法运算可以使用Py火花的内置函数以更有效的方式实现:

从pyspark.sql.函数导入col
Df = Df.withColumn("col1_double "， col("col1") * 2)

在无法使用内置函数执行操作并且需要Python UDF的情况下, 向量化UDF可以提供更有效的替代方案:

从pyspark.sql.函数导入p和as_udf
从pyspark.sql.types import IntegerType

@p和as_udf (IntegerType ())
Def multiply_by_two_pd(s: pd.Series) -> pd.系列:
   返回s * 2
Df = Df.withColumn(“col1_doubled multiply_by_two_pd (df(“col1”)))

这个方法应用函数 multiply_by_two_pd 一次处理整个数据系列，从而减少了序列化开销. 的输入和返回 multiply_by_two_pd 功能都是熊猫系列. P和as Series是一维标记数组，可用于表示DataFrame中单列中的数据.

优化数据处理性能

随着机器学习和大数据变得越来越普遍, 工程师们正在采用Apache火花来处理这些技术需要处理的大量数据. 提高火花的性能涉及一系列策略, 所有的设计都是为了优化可用资源的使用. 实现这里讨论的技术将有助于火花更有效地处理大量数据.