1+x数字营销网站,创建网站的英语,视频网站视频预览怎么做的,做的一个网站多少钱Pandas做数据处理可以说是yyds#xff01;而它的缺点也是非常明显#xff0c;Pandas 只能单机处理#xff0c;它不能随数据量线性伸缩。例如#xff0c;如果 pandas 试图读取的数据集大于一台机器的可用内存#xff0c;则会因内存不足而失败。
另外 pandas 在处理大型数据…Pandas做数据处理可以说是yyds而它的缺点也是非常明显Pandas 只能单机处理它不能随数据量线性伸缩。例如如果 pandas 试图读取的数据集大于一台机器的可用内存则会因内存不足而失败。
另外 pandas 在处理大型数据方面非常慢虽然有像Dask 或 Vaex 等其他库来优化提升数据处理速度但在大数据处理神之框架Spark面前也是小菜一碟。
幸运的是在新的 Spark 3.2 版本中出现了一个新的Pandas API将pandas大部分功能都集成到PySpark中使用pandas的接口就能使用Spark因为 Spark 上的 Pandas API 在后台使用 Spark这样就能达到强强联手的效果可以说是非常强大非常方便。
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 数据集的性能对比。
技术提升
本文由技术群粉丝分享项目源码、数据、技术交流提升均可加交流群获取群友已超过2000人添加时最好的备注方式为来源兴趣方向方便找到志同道合的朋友 方式①、添加微信号pythoner666备注来自CSDN 方式②、微信搜索公众号Python学习与数据挖掘后台回复加群 多线程和 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 int8
decimal object
float float32
double float64
integer int32
long int64
short int16
timestamp datetime64[ns]
string object
boolean bool
date object
dtype: objectPandas-on-Spark vs Spark 函数
在 Spark 中的 DataFrame 及其在 Pandas-on-Spark 中的最常用函数。注意Pandas-on-Spark 和 Pandas 在语法上的唯一区别就是 import pyspark.pandas as ps 一行。
当你看完如下内容后你会发现即使您不熟悉 Spark也可以通过 Pandas API 轻松使用。
导入库
# 运行Spark
from pyspark.sql import SparkSession
spark SparkSession.builder \ .appName(Spark) \ .getOrCreate()
# 在Spark上运行Pandas
import pyspark.pandas as ps 读取数据
以 old dog iris 数据集为例。
# SPARK
sdf spark.read.options(inferSchemaTrue, headerTrue).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()之间的主要区别在于前者需要返回相同长度的输入而后者不需要。
# transform
psdf ps.DataFrame({a: [1,2,3], b:[4,5,6]})
def pandas_plus(pser): return pser 1 # 应该总是返回与输入相同的长度。 psdf.transform(pandas_plus) # apply
psdf ps.DataFrame({a: [1,2,3], b:[5,6,7]})
def pandas_plus(pser): return pser[pser % 2 1] # 允许任意长度 psdf.apply(pandas_plus) 在这种情况下每个函数采用一个 pandas SeriesSpark 上的 pandas API 以分布式方式计算函数如下所示。 在“列”轴的情况下该函数将每一行作为一个熊猫系列。
psdf ps.DataFrame({a: [1,2,3], b:[4,5,6]})
def pandas_plus(pser): return sum(pser) # 允许任意长度
psdf.apply(pandas_plus, axiscolumns) 上面的示例将每一行的总和计算为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 ps
psdf 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 ps
psdf ps.DataFrame({a: [1, 2], A:[3, 4]}) Reference a is ambiguous, could be: a, a.;但可以在 Spark 配置spark.sql.caseSensitive中打开以启用它但需要自己承担风险。
from pyspark.sql import SparkSession
builder SparkSession.builder.appName(pandas-on-spark)
builder builder.config(spark.sql.caseSensitive, true)
builder.getOrCreate() import pyspark.pandas as ps
psdf ps.DataFrame({a: [1, 2], A:[3, 4]})
psdf a A
0 1 3
1 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 pd
data []
countries [London, New York, Helsinki]
pser pd.Series([20., 21., 12.], indexcountries)
for temperature in pser: assert temperature 0 if temperature 1000: temperature None data.append(temperature ** 2) pd.Series(data, indexcountries) London 400.0
New York 441.0
Helsinki 144.0
dtype: float64但是对于 Spark 上的 pandas API它的工作原理与上述相同。上面的示例也可以更改为直接使用 pandas-on-Spark API如下所示
import pyspark.pandas as ps
import numpy as np
countries [London, New York, Helsinki]
psser ps.Series([20., 21., 12.], indexcountries)
def square(temperature) - np.float64: assert temperature 0 if temperature 1000: temperature None return temperature ** 2 psser.apply(square) London 400.0
New York 441.0
Helsinki 144.0减少对不同 DataFrame 的操作
Spark 上的 Pandas API 默认不允许对不同 DataFrame或 Series进行操作以防止昂贵的操作。只要有可能就应该避免这种操作。
写在最后
到目前为止我们将能够在 Spark 上使用 Pandas。这将会导致Pandas 速度的大大提高迁移到 Spark 时学习曲线的减少以及单机计算和分布式计算在同一代码库中的合并。
参考资料
[1]PySpark 指南: https://spark.apache.org/docs/latest/api/python/user_guide/pandas_on_spark/pandas_pyspark.html
[2]PySpark 指南: https://spark.apache.org/docs/latest/api/python/user_guide/pandas_on_spark/types.html
[3]默认索引类型: https://spark.apache.org/docs/latest/api/python/user_guide/pandas_on_spark/options.html#default-index-type
