Pandas与PySpark强强联手，功能与速度齐飞
使用python做数据处理的数据科学家或数据从业者，对数据科学包pandas并不陌生，也不乏像主页君一样的pandas重度使用者，项目开始写的第一行代码，大多是 import pandas as pd。pandas做数据处理可以说是yyds！而他的缺点也是非常明显，pandas 只能单机处理，它不能随数据量线性伸缩。例如，如果 pandas 试图读取的数据集大于一台机器的可用内存，则会因内存不足而失败。
另外 pandas 在处理大型数据方面非常慢，虽然有像dask 或 vaex 等其他库来优化提升数据处理速度，但在大数据处理神之框架spark面前，也是小菜一碟。
幸运的是，在新的 spark 3.2 版本中，出现了一个新的pandas api，将pandas大部分功能都集成到pyspark中，使用pandas的接口，就能使用spark，因为 spark 上的 pandas api 在后台使用 spark，这样就能达到强强联手的效果，可以说是非常强大，非常方便。
这一切都始于 2019 年 spark + ai 峰会。koalas 是一个开源项目，可以在 spark 之上使用 pandas。一开始，它只覆盖了 pandas 的一小部分功能，但后来逐渐壮大起来。现在，在新的 spark 3.2 版本中，koalas 已合并到 pyspark。
spark 现在集成了 pandas api，因此可以在 spark 上运行 pandas。只需要更改一行代码：
import pyspark.pandas as ps 由此我们可以获得诸多的优势：
如果我们熟悉使用python 和 pandas，但不熟悉 spark，可以省略了需复杂的学习过程而立即使用pyspark。 可以为所有内容使用一个代码库：无论是小数据和大数据，还是单机和分布式机器。 可以在spark分布式框架上，更快地运行 pandas 代码。 最后一点尤其值得注意。
一方面，可以将分布式计算应用于在 pandas 中的代码。且借助 spark 引擎，代码即使在单台机器上也会更快！下图展示了在一台机器（具有 96 个 vcpu 和 384 gibs 内存）上运行 spark 和单独调用 pandas 分析 130gb 的 csv 数据集的性能对比。
多线程和 spark sql catalyst optimizer 都有助于优化性能。例如，join count 操作在整个阶段代码生成时快 4 倍：没有代码生成时为 5.9 秒，代码生成时为 1.6 秒。
spark 在链式操作（chaining operations）中具有特别显着的优势。catalyst 查询优化器可以识别过滤器以明智地过滤数据并可以应用基于磁盘的连接（disk-based joins），而 pandas 倾向于每一步将所有数据加载到内存中。
现在是不是迫不及待的想尝试如何在 spark 上使用 pandas api 编写一些代码？我们现在就开始吧！
在 pandas / pandas-on-spark / spark 之间切换
需要知道的第一件事是我们到底在使用什么。在使用 pandas 时，使用类pandas.core.frame.dataframe。在 spark 中使用 pandas api 时，使用pyspark.pandas.frame.dataframe。虽然两者相似，但不相同。主要区别在于前者在单机中，而后者是分布式的。
可以使用 pandas-on-spark 创建一个 dataframe 并将其转换为 pandas，反之亦然：
# import pandas-on-spark import pyspark.pandas as ps# 使用 pandas-on-spark 创建一个 dataframe ps_df = ps.dataframe(range(10))# 将 pandas-on-spark dataframe 转换为 pandas dataframe pd_df = ps_df.to_pandas()# 将 pandas dataframe 转换为 pandas-on-spark dataframe ps_df = ps.from_pandas(pd_df) 注意，如果使用多台机器，则在将 pandas-on-spark dataframe 转换为 pandas dataframe 时，数据会从多台机器传输到一台机器，反之亦然（可参阅pyspark 指南[1]）。
还可以将 pandas-on-spark dataframe 转换为 spark dataframe，反之亦然：
# 使用 pandas-on-spark 创建一个 dataframe ps_df = ps.dataframe(range(10))# 将 pandas-on-spark dataframe 转换为 spark dataframe spark_df = ps_df.to_spark()# 将 spark dataframe 转换为 pandas-on-spark dataframe ps_df_new = spark_df.to_pandas_on_spark() 数据类型如何改变？
在使用 pandas-on-spark 和 pandas 时，数据类型基本相同。将 pandas-on-spark dataframe 转换为 spark dataframe 时，数据类型会自动转换为适当的类型（请参阅pyspark 指南[2]）
下面的示例显示了在转换时是如何将数据类型从 pyspark dataframe 转换为 pandas-on-spark dataframe。
>>> sdf = spark.createdataframe([... (1, decimal(1.0), 1., 1., 1, 1, 1, datetime(2020, 10, 27), 1, true, datetime(2020, 10, 27)),... ], 'tinyint tinyint, decimal decimal, float float, double double, integer integer, long long, short short, timestamp timestamp, string string, boolean boolean, date date')>>> sdf dataframe[tinyint: tinyint, decimal: decimal(10,0),float: float, double: double, integer: int,long: bigint, short: smallint, timestamp: timestamp, string: string, boolean: boolean, date: date] psdf = sdf.pandas_api()psdf.dtypes tinyint int8decimal objectfloat float32double float64integer int32long int64short int16timestamp datetime64[ns]string objectboolean booldate objectdtype: object pandas-on-spark vs spark 函数
在 spark 中的 dataframe 及其在 pandas-on-spark 中的最常用函数。注意，pandas-on-spark 和 pandas 在语法上的唯一区别就是 import pyspark.pandas as ps 一行。
当你看完如下内容后，你会发现，即使您不熟悉 spark，也可以通过 pandas api 轻松使用。
导入库 # 运行sparkfrom pyspark.sql import sparksessionspark = sparksession.builder .appname(spark) .getorcreate()# 在spark上运行pandasimport pyspark.pandas as ps 读取数据 以 old dog iris 数据集为例。
# spark sdf = spark.read.options(inferschema='true', header='true').csv('iris.csv')# pandas-on-spark pdf = ps.read_csv('iris.csv') 选择 # spark sdf.select(sepal_length,sepal_width).show()# pandas-on-spark pdf[[sepal_length,sepal_width]].head() 删除列 # spark sdf.drop('sepal_length').show()# pandas-on-spark pdf.drop('sepal_length').head() 删除重复项 # spark sdf.dropduplicates([sepal_length,sepal_width]).show()# pandas-on-spark pdf[[sepal_length, sepal_width]].drop_duplicates() 筛选 # spark sdf.filter( (sdf.flower_type == iris-setosa) & (sdf.petal_length > 1.5) ).show()# pandas-on-spark pdf.loc[ (pdf.flower_type == iris-setosa) & (pdf.petal_length > 1.5) ].head() 计数 # spark sdf.filter(sdf.flower_type == iris-virginica).count()# pandas-on-spark pdf.loc[pdf.flower_type == iris-virginica].count() 唯一值 # spark sdf.select(flower_type).distinct().show()# pandas-on-spark pdf[flower_type].unique() 排序 # spark sdf.sort(sepal_length, sepal_width).show()# pandas-on-spark pdf.sort_values([sepal_length, sepal_width]).head() 分组 # spark sdf.groupby(flower_type).count().show()# pandas-on-spark pdf.groupby(flower_type).count() 替换 # spark sdf.replace(iris-setosa, setosa).show()# pandas-on-spark pdf.replace(iris-setosa, setosa).head() 连接 #spark sdf.union(sdf)# pandas-on-spark pdf.append(pdf) transform 和 apply 函数应用
有许多 api 允许用户针对 pandas-on-spark dataframe 应用函数，例如：
dataframe.transform() dataframe.apply()dataframe.pandas_on_spark.transform_batch() dataframe.pandas_on_spark.apply_batch() series.pandas_on_spark.transform_batch() 每个 api 都有不同的用途，并且在内部工作方式不同。
transform 和 apply dataframe.transform()和dataframe.apply()之间的主要区别在于，前者需要返回相同长度的输入，而后者不需要。
# transformpsdf = ps.dataframe({'a': [1,2,3], 'b':[4,5,6]})def pandas_plus(pser): return pser + 1 # 应该总是返回与输入相同的长度。psdf.transform(pandas_plus)# applypsdf = ps.dataframe({'a': [1,2,3], 'b':[5,6,7]})def pandas_plus(pser): return pser[pser % 2 == 1] # 允许任意长度psdf.apply(pandas_plus) 在这种情况下，每个函数采用一个 pandas series，spark 上的 pandas api 以分布式方式计算函数，如下所示。
在“列”轴的情况下，该函数将每一行作为一个熊猫系列。
psdf = ps.dataframe({'a': [1,2,3], 'b':[4,5,6]})def pandas_plus(pser): return sum(pser) # 允许任意长度psdf.apply(pandas_plus, axis='columns') 上面的示例将每一行的总和计算为pands series
pandas_on_spark.transform_batch和pandas_on_spark.apply_batch batch 后缀表示 pandas-on-spark dataframe 或 series 中的每个块。api 对 pandas-on-spark dataframe 或 series 进行切片，然后以 pandas dataframe 或 series 作为输入和输出应用给定函数。请参阅以下示例：
psdf = ps.dataframe({'a': [1,2,3], 'b':[4,5,6]})def pandas_plus(pdf): return pdf + 1 # 应该总是返回与输入相同的长度。psdf.pandas_on_spark.transform_batch(pandas_plus)psdf = ps.dataframe({'a': [1,2,3], 'b':[4,5,6]})def pandas_plus(pdf): return pdf[pdf.a > 1] # 允许任意长度psdf.pandas_on_spark.apply_batch(pandas_plus) 两个示例中的函数都将 pandas dataframe 作为 pandas-on-spark dataframe 的一个块，并输出一个 pandas dataframe。spark 上的 pandas api 将 pandas 数据帧组合为 pandas-on-spark 数据帧。
在 spark 上使用 pandas api的注意事项 避免shuffle 某些操作，例如sort_values在并行或分布式环境中比在单台机器上的内存中更难完成，因为它需要将数据发送到其他节点，并通过网络在多个节点之间交换数据。
避免在单个分区上计算 另一种常见情况是在单个分区上进行计算。目前， dataframe.rank 等一些 api 使用 pyspark 的 window 而不指定分区规范。这会将所有数据移动到单个机器中的单个分区中，并可能导致严重的性能下降。对于非常大的数据集，应避免使用此类 api。
不要使用重复的列名 不允许使用重复的列名，因为 spark sql 通常不允许这样做。spark 上的 pandas api 继承了这种行为。例如，见下文：
import pyspark.pandas as pspsdf = ps.dataframe({'a': [1, 2], 'b':[3, 4]})psdf.columns = [a, a] reference 'a' is ambiguous, could be: a, a.; 此外，强烈建议不要使用区分大小写的列名。spark 上的 pandas api 默认不允许它。
import pyspark.pandas as pspsdf = ps.dataframe({'a': [1, 2], 'a':[3, 4]}) reference 'a' is ambiguous, could be: a, a.; 但可以在 spark 配置spark.sql.casesensitive中打开以启用它，但需要自己承担风险。
from pyspark.sql import sparksessionbuilder = sparksession.builder.appname(pandas-on-spark)builder = builder.config(spark.sql.casesensitive, true)builder.getorcreate()import pyspark.pandas as pspsdf = ps.dataframe({'a': [1, 2], 'a':[3, 4]})psdf a a0 1 31 2 4 使用默认索引 pandas-on-spark 用户面临的一个常见问题是默认索引导致性能下降。当索引未知时，spark 上的 pandas api 会附加一个默认索引，例如 spark dataframe 直接转换为 pandas-on-spark dataframe。
如果计划在生产中处理大数据，请通过将默认索引配置为distributed或distributed-sequence来使其确保为分布式。
有关配置默认索引的更多详细信息，请参阅默认索引类型[3]。
在 spark 上使用 pandas api 尽管 spark 上的 pandas api 具有大部分与 pandas 等效的 api，但仍有一些 api 尚未实现或明确不受支持。因此尽可能直接在 spark 上使用 pandas api。
例如，spark 上的 pandas api 没有实现__iter__()，阻止用户将所有数据从整个集群收集到客户端（驱动程序）端。不幸的是，许多外部 api，例如 min、max、sum 等 python 的内置函数，都要求给定参数是可迭代的。对于 pandas，它开箱即用，如下所示：
>>> import pandas as pd>>> max(pd.series([1, 2, 3]))3>>> min(pd.series([1, 2, 3]))1>>> sum(pd.series([1, 2, 3]))6 pandas 数据集存在于单台机器中，自然可以在同一台机器内进行本地迭代。但是，pandas-on-spark 数据集存在于多台机器上，并且它们是以分布式方式计算的。很难在本地迭代，很可能用户在不知情的情况下将整个数据收集到客户端。因此，最好坚持使用 pandas-on-spark api。上面的例子可以转换如下：
>>> import pyspark.pandas as ps>>> ps.series([1, 2, 3]).max()3>>> ps.series([1, 2, 3]).min()1>>> ps.series([1, 2, 3]).sum()6 pandas 用户的另一个常见模式可能是依赖列表推导式或生成器表达式。但是，它还假设数据集在引擎盖下是本地可迭代的。因此，它可以在 pandas 中无缝运行，如下所示：
import pandas as pddata = []countries = ['london', 'new york', 'helsinki']pser = pd.series([20., 21., 12.], index=countries)for temperature in pser: assert temperature > 0 if temperature > 1000: temperature = none data.append(temperature ** 2)pd.series(data, index=countries) london 400.0new york 441.0helsinki 144.0dtype: float64 但是，对于 spark 上的 pandas api，它的工作原理与上述相同。上面的示例也可以更改为直接使用 pandas-on-spark api，如下所示：
import pyspark.pandas as psimport numpy as npcountries = ['london', 'new york', 'helsinki']psser = ps.series([20., 21., 12.], index=countries)def square(temperature) -> np.float64: assert temperature > 0 if temperature > 1000: temperature = none return temperature ** 2psser.apply(square) london 400.0new york 441.0helsinki 144.0 减少对不同 dataframe 的操作 spark 上的 pandas api 默认不允许对不同 dataframe（或 series）进行操作，以防止昂贵的操作。只要有可能，就应该避免这种操作。
写在最后
到目前为止，我们将能够在 spark 上使用 pandas。这将会导致pandas 速度的大大提高，迁移到 spark 时学习曲线的减少，以及单机计算和分布式计算在同一代码库中的合并。
哪一些原因让大家不想去了解区块链
Python中 Map/Filter/Reduce的差异
AI新突破将会带来什么好处
华为mate10Pro和小米mix2对比评测,哪个更值得入手?
FP7126高精度调光线路原理图
Pandas与PySpark强强联手，功能与速度齐飞
真实避障能力有几分？那些让扫地机器人“汗流浃背”的障碍物
谈P2P电影服务器(最经济方案)
arp攻击原理_arp攻击怎么解决
全新ALIENWARE X系列游戏本发布：惟精惟一 卓尔不群
igct和igbt有什么区别
uasb反应器的特点
利用i.MXRT1xxx系列ROM集成的DCD功能可轻松配置指定外设
一文了解MCU单片机的定时器功能
网络端口大全
欧姆龙CP1/CJ1/CJ2/CS1系列PLC联网方案（编程+监控）
三星拟向其他厂商出售Exynos处理器，来抢占全球手机处理器市场份额
直流电桥定义和使用方法详解以及如何使用电桥测量电容，电阻，电感
基于NIVIDA RTX的GPU加速边缘AI服务器——AIR-500D
汽车线束检测设备检测哪些项目内容
另外 pandas 在处理大型数据方面非常慢，虽然有像dask 或 vaex 等其他库来优化提升数据处理速度，但在大数据处理神之框架spark面前，也是小菜一碟。
幸运的是，在新的 spark 3.2 版本中，出现了一个新的pandas api，将pandas大部分功能都集成到pyspark中，使用pandas的接口，就能使用spark，因为 spark 上的 pandas api 在后台使用 spark，这样就能达到强强联手的效果，可以说是非常强大，非常方便。
这一切都始于 2019 年 spark + ai 峰会。koalas 是一个开源项目，可以在 spark 之上使用 pandas。一开始，它只覆盖了 pandas 的一小部分功能，但后来逐渐壮大起来。现在，在新的 spark 3.2 版本中，koalas 已合并到 pyspark。
spark 现在集成了 pandas api，因此可以在 spark 上运行 pandas。只需要更改一行代码：
import pyspark.pandas as ps 由此我们可以获得诸多的优势：
如果我们熟悉使用python 和 pandas，但不熟悉 spark，可以省略了需复杂的学习过程而立即使用pyspark。 可以为所有内容使用一个代码库：无论是小数据和大数据，还是单机和分布式机器。 可以在spark分布式框架上，更快地运行 pandas 代码。 最后一点尤其值得注意。
一方面，可以将分布式计算应用于在 pandas 中的代码。且借助 spark 引擎，代码即使在单台机器上也会更快！下图展示了在一台机器（具有 96 个 vcpu 和 384 gibs 内存）上运行 spark 和单独调用 pandas 分析 130gb 的 csv 数据集的性能对比。
多线程和 spark sql catalyst optimizer 都有助于优化性能。例如，join count 操作在整个阶段代码生成时快 4 倍：没有代码生成时为 5.9 秒，代码生成时为 1.6 秒。
spark 在链式操作（chaining operations）中具有特别显着的优势。catalyst 查询优化器可以识别过滤器以明智地过滤数据并可以应用基于磁盘的连接（disk-based joins），而 pandas 倾向于每一步将所有数据加载到内存中。
现在是不是迫不及待的想尝试如何在 spark 上使用 pandas api 编写一些代码？我们现在就开始吧！
在 pandas / pandas-on-spark / spark 之间切换
需要知道的第一件事是我们到底在使用什么。在使用 pandas 时，使用类pandas.core.frame.dataframe。在 spark 中使用 pandas api 时，使用pyspark.pandas.frame.dataframe。虽然两者相似，但不相同。主要区别在于前者在单机中，而后者是分布式的。
可以使用 pandas-on-spark 创建一个 dataframe 并将其转换为 pandas，反之亦然：
# import pandas-on-spark import pyspark.pandas as ps# 使用 pandas-on-spark 创建一个 dataframe ps_df = ps.dataframe(range(10))# 将 pandas-on-spark dataframe 转换为 pandas dataframe pd_df = ps_df.to_pandas()# 将 pandas dataframe 转换为 pandas-on-spark dataframe ps_df = ps.from_pandas(pd_df) 注意，如果使用多台机器，则在将 pandas-on-spark dataframe 转换为 pandas dataframe 时，数据会从多台机器传输到一台机器，反之亦然（可参阅pyspark 指南[1]）。
还可以将 pandas-on-spark dataframe 转换为 spark dataframe，反之亦然：
# 使用 pandas-on-spark 创建一个 dataframe ps_df = ps.dataframe(range(10))# 将 pandas-on-spark dataframe 转换为 spark dataframe spark_df = ps_df.to_spark()# 将 spark dataframe 转换为 pandas-on-spark dataframe ps_df_new = spark_df.to_pandas_on_spark() 数据类型如何改变？
在使用 pandas-on-spark 和 pandas 时，数据类型基本相同。将 pandas-on-spark dataframe 转换为 spark dataframe 时，数据类型会自动转换为适当的类型（请参阅pyspark 指南[2]）
下面的示例显示了在转换时是如何将数据类型从 pyspark dataframe 转换为 pandas-on-spark dataframe。
>>> sdf = spark.createdataframe([... (1, decimal(1.0), 1., 1., 1, 1, 1, datetime(2020, 10, 27), 1, true, datetime(2020, 10, 27)),... ], 'tinyint tinyint, decimal decimal, float float, double double, integer integer, long long, short short, timestamp timestamp, string string, boolean boolean, date date')>>> sdf dataframe[tinyint: tinyint, decimal: decimal(10,0),float: float, double: double, integer: int,long: bigint, short: smallint, timestamp: timestamp, string: string, boolean: boolean, date: date] psdf = sdf.pandas_api()psdf.dtypes tinyint int8decimal objectfloat float32double float64integer int32long int64short int16timestamp datetime64[ns]string objectboolean booldate objectdtype: object pandas-on-spark vs spark 函数
在 spark 中的 dataframe 及其在 pandas-on-spark 中的最常用函数。注意，pandas-on-spark 和 pandas 在语法上的唯一区别就是 import pyspark.pandas as ps 一行。
当你看完如下内容后，你会发现，即使您不熟悉 spark，也可以通过 pandas api 轻松使用。
导入库 # 运行sparkfrom pyspark.sql import sparksessionspark = sparksession.builder .appname(spark) .getorcreate()# 在spark上运行pandasimport pyspark.pandas as ps 读取数据 以 old dog iris 数据集为例。
# spark sdf = spark.read.options(inferschema='true', header='true').csv('iris.csv')# pandas-on-spark pdf = ps.read_csv('iris.csv') 选择 # spark sdf.select(sepal_length,sepal_width).show()# pandas-on-spark pdf[[sepal_length,sepal_width]].head() 删除列 # spark sdf.drop('sepal_length').show()# pandas-on-spark pdf.drop('sepal_length').head() 删除重复项 # spark sdf.dropduplicates([sepal_length,sepal_width]).show()# pandas-on-spark pdf[[sepal_length, sepal_width]].drop_duplicates() 筛选 # spark sdf.filter( (sdf.flower_type == iris-setosa) & (sdf.petal_length > 1.5) ).show()# pandas-on-spark pdf.loc[ (pdf.flower_type == iris-setosa) & (pdf.petal_length > 1.5) ].head() 计数 # spark sdf.filter(sdf.flower_type == iris-virginica).count()# pandas-on-spark pdf.loc[pdf.flower_type == iris-virginica].count() 唯一值 # spark sdf.select(flower_type).distinct().show()# pandas-on-spark pdf[flower_type].unique() 排序 # spark sdf.sort(sepal_length, sepal_width).show()# pandas-on-spark pdf.sort_values([sepal_length, sepal_width]).head() 分组 # spark sdf.groupby(flower_type).count().show()# pandas-on-spark pdf.groupby(flower_type).count() 替换 # spark sdf.replace(iris-setosa, setosa).show()# pandas-on-spark pdf.replace(iris-setosa, setosa).head() 连接 #spark sdf.union(sdf)# pandas-on-spark pdf.append(pdf) transform 和 apply 函数应用
有许多 api 允许用户针对 pandas-on-spark dataframe 应用函数，例如：
dataframe.transform() dataframe.apply()dataframe.pandas_on_spark.transform_batch() dataframe.pandas_on_spark.apply_batch() series.pandas_on_spark.transform_batch() 每个 api 都有不同的用途，并且在内部工作方式不同。
transform 和 apply dataframe.transform()和dataframe.apply()之间的主要区别在于，前者需要返回相同长度的输入，而后者不需要。
# transformpsdf = ps.dataframe({'a': [1,2,3], 'b':[4,5,6]})def pandas_plus(pser): return pser + 1 # 应该总是返回与输入相同的长度。psdf.transform(pandas_plus)# applypsdf = ps.dataframe({'a': [1,2,3], 'b':[5,6,7]})def pandas_plus(pser): return pser[pser % 2 == 1] # 允许任意长度psdf.apply(pandas_plus) 在这种情况下，每个函数采用一个 pandas series，spark 上的 pandas api 以分布式方式计算函数，如下所示。
在“列”轴的情况下，该函数将每一行作为一个熊猫系列。
psdf = ps.dataframe({'a': [1,2,3], 'b':[4,5,6]})def pandas_plus(pser): return sum(pser) # 允许任意长度psdf.apply(pandas_plus, axis='columns') 上面的示例将每一行的总和计算为pands series
pandas_on_spark.transform_batch和pandas_on_spark.apply_batch batch 后缀表示 pandas-on-spark dataframe 或 series 中的每个块。api 对 pandas-on-spark dataframe 或 series 进行切片，然后以 pandas dataframe 或 series 作为输入和输出应用给定函数。请参阅以下示例：
psdf = ps.dataframe({'a': [1,2,3], 'b':[4,5,6]})def pandas_plus(pdf): return pdf + 1 # 应该总是返回与输入相同的长度。psdf.pandas_on_spark.transform_batch(pandas_plus)psdf = ps.dataframe({'a': [1,2,3], 'b':[4,5,6]})def pandas_plus(pdf): return pdf[pdf.a > 1] # 允许任意长度psdf.pandas_on_spark.apply_batch(pandas_plus) 两个示例中的函数都将 pandas dataframe 作为 pandas-on-spark dataframe 的一个块，并输出一个 pandas dataframe。spark 上的 pandas api 将 pandas 数据帧组合为 pandas-on-spark 数据帧。
在 spark 上使用 pandas api的注意事项 避免shuffle 某些操作，例如sort_values在并行或分布式环境中比在单台机器上的内存中更难完成，因为它需要将数据发送到其他节点，并通过网络在多个节点之间交换数据。
避免在单个分区上计算 另一种常见情况是在单个分区上进行计算。目前， dataframe.rank 等一些 api 使用 pyspark 的 window 而不指定分区规范。这会将所有数据移动到单个机器中的单个分区中，并可能导致严重的性能下降。对于非常大的数据集，应避免使用此类 api。
不要使用重复的列名 不允许使用重复的列名，因为 spark sql 通常不允许这样做。spark 上的 pandas api 继承了这种行为。例如，见下文：
import pyspark.pandas as pspsdf = ps.dataframe({'a': [1, 2], 'b':[3, 4]})psdf.columns = [a, a] reference 'a' is ambiguous, could be: a, a.; 此外，强烈建议不要使用区分大小写的列名。spark 上的 pandas api 默认不允许它。
import pyspark.pandas as pspsdf = ps.dataframe({'a': [1, 2], 'a':[3, 4]}) reference 'a' is ambiguous, could be: a, a.; 但可以在 spark 配置spark.sql.casesensitive中打开以启用它，但需要自己承担风险。
from pyspark.sql import sparksessionbuilder = sparksession.builder.appname(pandas-on-spark)builder = builder.config(spark.sql.casesensitive, true)builder.getorcreate()import pyspark.pandas as pspsdf = ps.dataframe({'a': [1, 2], 'a':[3, 4]})psdf a a0 1 31 2 4 使用默认索引 pandas-on-spark 用户面临的一个常见问题是默认索引导致性能下降。当索引未知时，spark 上的 pandas api 会附加一个默认索引，例如 spark dataframe 直接转换为 pandas-on-spark dataframe。
如果计划在生产中处理大数据，请通过将默认索引配置为distributed或distributed-sequence来使其确保为分布式。
有关配置默认索引的更多详细信息，请参阅默认索引类型[3]。
在 spark 上使用 pandas api 尽管 spark 上的 pandas api 具有大部分与 pandas 等效的 api，但仍有一些 api 尚未实现或明确不受支持。因此尽可能直接在 spark 上使用 pandas api。
例如，spark 上的 pandas api 没有实现__iter__()，阻止用户将所有数据从整个集群收集到客户端（驱动程序）端。不幸的是，许多外部 api，例如 min、max、sum 等 python 的内置函数，都要求给定参数是可迭代的。对于 pandas，它开箱即用，如下所示：
>>> import pandas as pd>>> max(pd.series([1, 2, 3]))3>>> min(pd.series([1, 2, 3]))1>>> sum(pd.series([1, 2, 3]))6 pandas 数据集存在于单台机器中，自然可以在同一台机器内进行本地迭代。但是，pandas-on-spark 数据集存在于多台机器上，并且它们是以分布式方式计算的。很难在本地迭代，很可能用户在不知情的情况下将整个数据收集到客户端。因此，最好坚持使用 pandas-on-spark api。上面的例子可以转换如下：
>>> import pyspark.pandas as ps>>> ps.series([1, 2, 3]).max()3>>> ps.series([1, 2, 3]).min()1>>> ps.series([1, 2, 3]).sum()6 pandas 用户的另一个常见模式可能是依赖列表推导式或生成器表达式。但是，它还假设数据集在引擎盖下是本地可迭代的。因此，它可以在 pandas 中无缝运行，如下所示：
import pandas as pddata = []countries = ['london', 'new york', 'helsinki']pser = pd.series([20., 21., 12.], index=countries)for temperature in pser: assert temperature > 0 if temperature > 1000: temperature = none data.append(temperature ** 2)pd.series(data, index=countries) london 400.0new york 441.0helsinki 144.0dtype: float64 但是，对于 spark 上的 pandas api，它的工作原理与上述相同。上面的示例也可以更改为直接使用 pandas-on-spark api，如下所示：
import pyspark.pandas as psimport numpy as npcountries = ['london', 'new york', 'helsinki']psser = ps.series([20., 21., 12.], index=countries)def square(temperature) -> np.float64: assert temperature > 0 if temperature > 1000: temperature = none return temperature ** 2psser.apply(square) london 400.0new york 441.0helsinki 144.0 减少对不同 dataframe 的操作 spark 上的 pandas api 默认不允许对不同 dataframe（或 series）进行操作，以防止昂贵的操作。只要有可能，就应该避免这种操作。
写在最后
到目前为止，我们将能够在 spark 上使用 pandas。这将会导致pandas 速度的大大提高，迁移到 spark 时学习曲线的减少，以及单机计算和分布式计算在同一代码库中的合并。
哪一些原因让大家不想去了解区块链
Python中 Map/Filter/Reduce的差异
AI新突破将会带来什么好处
华为mate10Pro和小米mix2对比评测,哪个更值得入手?
FP7126高精度调光线路原理图
Pandas与PySpark强强联手，功能与速度齐飞
真实避障能力有几分？那些让扫地机器人“汗流浃背”的障碍物
谈P2P电影服务器(最经济方案)
arp攻击原理_arp攻击怎么解决
全新ALIENWARE X系列游戏本发布：惟精惟一 卓尔不群
igct和igbt有什么区别
uasb反应器的特点
利用i.MXRT1xxx系列ROM集成的DCD功能可轻松配置指定外设
一文了解MCU单片机的定时器功能
网络端口大全
欧姆龙CP1/CJ1/CJ2/CS1系列PLC联网方案（编程+监控）
三星拟向其他厂商出售Exynos处理器，来抢占全球手机处理器市场份额
直流电桥定义和使用方法详解以及如何使用电桥测量电容，电阻，电感
基于NIVIDA RTX的GPU加速边缘AI服务器——AIR-500D
汽车线束检测设备检测哪些项目内容