PySpark大数据技术与应用 课件 第3章 基于PySpark的DataFrame操作

SparkSQL概述1SparkSQL主要功能目录SparkSQL起源与发展历程2SparkSQL数据核心抽象DataFrame3SparkSQL是Spark用于处理结构化数据的一个组件，其编程抽象为DataFrame。作为分布式SQL查询引擎，SparkSQL允许用户通过SQL、DataFrameAPI和DataSetAPI这3种方式处理结构化数据。SparkSQL将关系处理与Spark函数编程相结合，并支持读取多种数据源的数据，如CSV、JSON、JDBC、Hive等。Spark

SQL概述SparkSQL的起源与发展紧密相连于大数据处理技术的进步和需求，其前身是Shark，后来为了支持更广泛的数据源和优化性能，发展为全新的SparkSQL项目。SparkSQL的起源可以追溯到Hive。Hive作为Hadoop生态系统中的组件，主要服务于熟悉SQL但不熟悉MapReduce编程模型的用户，它允许数据分析师使用SQL语句来操作HDFS上的数据。然而，Hive在执行过程中存在大量的中间磁盘落地过程，导致效率低下。为了提高SQL-on-Hadoop的效率，Shark作为响应此需求的解决方案之一被提出。Shark修改了Hive架构中的内存管理、物理计划、执行模块，并运行在Spark引擎上，使得SQL查询速度得到显著提升。Spark

SQL的起源与发展随着技术的发展，Shark对Hive的过度依赖成为了限制Spark发展的瓶颈。因此，Spark团队决定抛弃Shark的代码，转而开发全新的SparkSQL项目。SparkSQL不仅摆脱了对Hive的依赖，还支持多种数据源，如Hive、Parquet、JSON、JDBC等，同时采用了多种性能优化技术，比如内存列存储、字节码生成和动态评估的成本模型等。从历史的角度来看，SparkSQL的出现和发展体现了开源社区对于技术迭代的能力，即不断推陈出新，以适应不断变化的数据处理需求。如今，SparkSQL已经成为ApacheSpark生态中最为活跃的子项目之一，它的重要性伴随着ApacheSpark的发展而日益突显。Spark

SQL的起源与发展1SparkSQL主要功能目录SparkSQL起源与发展历程2SparkSQL数据核心抽象DataFrame3SparkSQL允许开发人员通过SQL语句简单快捷地操作海量结构化数据，进行大数据分析。多数据源支持：从各种结构化文件和数据库中读取数据进行高效分析。广泛的连接性：提供JDBC和ODBC接口，不局限于Spark程序内查询。紧密集成的分析能力：将SQL查询与RDD结合，提供跨语言API支持复杂分析。模块化协作：与Streaming、MLlib、GraphX等模块紧密结合处理结构化数据。SparkSQL主要功能1SparkSQL主要功能目录SparkSQL起源与发展历程2SparkSQL数据核心抽象DataFrame3DataFrame：

SparkSQL的核心数据抽象，结合了RDD的弹性与SQL的操作便捷性定义与结构：带模式的RDD（SchemaRDD），提供名称和类型信息创建方式：支持从结构化文件、数据库、Hive表等多源创建API支持：Scala、Java、Python、R语言的API兼容性高级特性：自Spark1.6起，DataSetAPI为DataFrame带来类型安全扩展SparkSQL数据核心抽象DataFrame1.DataFrame概念NameAgeHeightStringIntDoubleStringIntDoubleStringIntDoubleDataFrame和RDD的共同特征不可变性：一旦创建，DataFrame和RDD中的数据不能被修改。如进行任何变更，会生成一个新的数据集。内存运行：在内存中执行计算，这样可以显著提高处理速度。弹性分布式计算能力：利用Spark的弹性分布式数据集的特性来处理大规模数据。SparkSQL数据核心抽象DataFrame2.

DataFrame与RDD比较DataFrame与RDD的主要区别数据结构：DataFrame提供了一种高层次的抽象，类似于传统数据库中的二维表格，具有行和列。而RDD是一个分布式的弹性数据集，可以看做是一个不可变的分布式对象集合。数据处理：DataFrame专注于结构化数据，RDD适用于各类数据源，尤其是非结构化数据。性能优化：DataFrame可以利用其结构化特性进行性能优化，比如执行代码生成和列式存储等，这些在RDD中不常见。API支持：DataFrame提供了更丰富的SQL类操作接口，而RDD的API更偏向于低层次的操作。SparkSQL数据核心抽象DataFrameDataFrame与RDD虽然都是Spark中重要的数据结构，但DataFrame通过提供结构化视图，使得数据处理更加直观和高效，而RDD则在处理非结构化数据时展现出更大的灵活性。SparkSQL数据核心抽象DataFrameDataFrame和DataSet都是SparkSQL中的重要数据结构，它们在类型系统、API支持和功能定位上存在一些差异。类型系统：DataFrame是DataSet[Row]的特例，它的每一行数据类型固定为Row。DataFrame通常只知道列字段，而不明确每一列的数据类型，在编译时不做类型检查。而DataSet是一个强类型的数据结构，用户需要定义具体的类（如Car类、Person类），DataSet知道每一列的名称和数据类型，提供了更严格的错误检查机制。SparkSQL数据核心抽象DataFrame3.

DataFrame与DataSet的比较API支持：DataFrame提供了丰富的SQL类操作接口，可以方便地进行数据选择、过滤、聚合等操作。DataSet则提供了更为丰富的API，包括了DataFrame的API以及更多针对特定类型的操作，这使得DataSet能够提供更高层次的抽象和更强的类型安全性。功能定位：DataFrame主要用于结构化数据处理，它可以看作是传统数据库中二维表格的一种分布式实现。而DataSet则结合了RDD和DataFrame的优点，不仅提供了结构化数据的处理能力，还允许用户自定义数据类型，从而可以进行更加复杂的数据处理和分析。总结来说，DataFrame和DataSet虽然在SparkSQL中都扮演着重要的角色，但DataFrame更适合那些不需要明确类型信息的通用数据处理任务，而DataSet则适用于需要强类型和复杂业务逻辑处理的场景。SparkSQL数据核心抽象DataFrameDataFrame为开发人员提供了更高级别的数据抽象，使得处理大型数据集更加容易。DataFrame优点主要体现在以下两个方面。性能优势：由于DataFrame引入了Schema，即数据结构的描述信息，因此后期Spark程序中的大量对象在进行网络传输时，只需针对数据的内容进行序列化，而无需传输数据结构信息。这样可以减少数据传输量，降低序列化和反序列的性能开销，弥补了RDD在这方面的不足。减少内存垃圾回收：DataFrame利用了堆外内存（Off-heap），直接在操作系统层上构建对象，而不是使用堆中的内存。这样可以节省堆内存空间，减少垃圾回收（GarbageCollection，GC）的频率，提高程序运行效率，弥补了RDD在堆内存占用和垃圾回收方面的不足。

SparkSQL数据核心抽象DataFrame4.

DataFrame的优点DataFrame在编译时不会进行类型检查，因而无法在编译时发现错误，只有在运行时才会发现错误。这可能导致在开发过程中难以及时发现和修复错误。DataFrame的API不是面向对象编程风格，不能直接通过对象调用方法操作数据，这限制了开发人员使用DataFrame的灵活性和可扩展性。SparkSQL数据核心抽象DataFrame5.

DataFrame的缺点pyspark.sql模块1pyspark.sql模块核心类目录pyspark.sql模块简介2pyspark.sql模块是PySpark用于处理结构化数据的模块，是SparkSQL面向Python的API。在Python环境中可以通过调用pyspark.sql模块中多个功能不同的类对Spark进行操作，完成大数据框架下的数据分析与处理。Spark的模块化设计思想能够使问题求解逻辑更清晰、使程序易于维护和修改。我们在日常生活、工作中遇到问题时，需发扬斗争精神，可将大问题分解为一个个小任务，逐个攻克，在团队中能更好地分工协作、各行其责。pyspark.sql模块pyspark.sql模块包含一系列类，各种类具有不同功能，与SparkSQL和DataFrame相关的类功能说明如下表。pyspark.sql模块简介类名功能说明pyspark.sql.SQLContextDataFrame和SQL功能的主入口，用于Spark1.xpyspark.sql.SparkSessionDataFrame和SQL功能的主入口，用于Spark2.x及以上版本pyspark.sql.DataFrame可将分布式数据集分组至指定列名的数据框中pyspark.sql.ColumnDataFrame中的列pyspark.sql.RowDataFrame中的行pyspark.sql.HiveContext访问Hive数据的主入口pyspark.sql.GroupedData由DataFrame.groupBy()方法创建的聚合方法集pyspark.sql.DataFrameNaFunctions处理丢失数据（NULL值）的功能pyspark.sql.DataFrameStatFunctionsDataFrame的统计功能pyspark.sql.functions包含DataFrame可用的内置函数pyspark.sql.types表示可用的数据类型列表pyspark.sql.Window用于处理窗口函数1pyspark.sql模块核心类目录pyspark.sql模块简介2pyspark.sql模块核心类主要有pyspark.sql.SparkSession和pyspark.sql.DataFrame。1．pyspark.sql.SparkSessionpyspark.sql.SparkSession是使用SparkSQL开发应用程序的主入口，SparkSQL编程从创建SparkSession对象开始。SparkSession对象不仅提供一系列API，如创建DataFrame对象API、读取外部数据源并转化DataFrame对象API、执行SQL查询和处理API，还负责Spark集群运行的控制、调优参数，是SparkSQL的Context环境和运行基础。pyspark.sql模块核心类在开发SparkSQL应用程序时，使用SparkSession.builder()函数创建SparkSession对象。首先从pyspark.sql模块导入SparkSession类。然后，使用SparkSession.builder来构建一个SparkSession对象。通过调用.appName(“YourAppName”)可以设置应用程序的名称。而.config("spark.some.config.option","some-value")用于设置一些配置选项。最后，通过调用.getOrCreate()方法来获取或创建一个SparkSession对象。请注意，需要将"YourAppName"替换为你自己的应用程序名称，并根据需要添加其他配置选项。pyspark.sql模块核心类pyspark.sql.DataFrame是PySpark中的一个核心类，用于表示分布式数据集合。它提供了类似于关系型数据库表的接口，可以执行各种数据操作和转换。pyspark.sql.DataFrame相关API可实现对DataFrame的多种操作，包括DataFrame的创建、查询、汇总、排序和保存等多种操作，这些DataFrame相关API功能强大、使用灵活且底层自带优化。DataFrame将SQL的select语句的各个组成部分封装为同名API，通过诸如select()、where()、orderBy()、groupBy()等DataFrameAPI灵活地组合实现SQL一样的逻辑表述。DataFrame编程像SQL一样，只需对计算条件、计算需求、最终结果进行声明式描述，不需要像RDD编程那样一步一步地对数据集进行原始操作。pyspark.sql模块核心类2．pyspark.sql.DataFrameSparkSQL编程归根结底是对RDD、DataFrame和DataSet的一系列操作，对DataFrame的一般操作步骤如下。创建DataFrame。利用SparkSession对象读取来自不同数据源的结构化数据创建DataFrame，或由RDD转化为DataFrame。在转化为DataFrame的过程中，需自识别或指定DataFrame的Schema。直接通过DataFrame的API进行数据分析或将DataFrame注册为表再对表数据进行查询，若将DataFrame注册为表，则需进行第（3）步。利用SparkSession提供的SQL方法在已注册的表上进行SQL查询，DataFrame在转化为临时视图时需根据实际情况选择是否转化为全局临时表。保存或删除DataFrame数据。pyspark.sql模块核心类DataFrame基础操作1DataFrame操作目录创建DataFrame对象2DataFrame输出操作3PySpark中的DataFrame等价于SparkSQL中的关系表。在PySpark中，DataFrame由列（Column）和行（Row）构成，DataFrame操作主要有创建、查询和输出操作，由一系列DataFrameAPI实现。本节示例操作数据集来自某航空公司客户管理系统，为了便于演示，将数据集分解为三个数据文件。air_data_customer.csv文件（会员基本信息表）air_data_flight.csv文件（乘机信息表）air_data_points.csv文件（积分信息表）DataFrame基础操作数据集概况会员基本信息表DataFrame基础操作文件名字段名说明air_data_customer.csvMEMBER_NO会员卡号FFP_DATE入会时间GENDER性别FFP_TIER 会员卡级别WORK_PROVINCE工作地WORK_COUNTRY工作地所属国家AGE年龄数据集概况乘机信息表DataFrame基础操作文件名字段名说明air_data_flight.csvMEMBER_NO会员卡号FIRST_FLIGHT_DATE第一次飞行日期FLIGHT_COUNT观测窗口内的飞行次数/次LOAD_TIME观测窗口的结束时间SEG_KM_SUM观测窗口的总飞行里程/公里LAST_FLIGHT_DATE末次飞行日期SUM_YR_1观测窗口中第一年的票价收入SUM_YR_2观测窗口中第二年的票价收入LAST_TO_END最后一次乘机时间至观测窗口结束时长/天AVG_DISCOUNT平均折扣率AVG_INTERVAL平均乘机时间间隔MAX_INTERVAL最大乘机间隔数据集概况积分信息表DataFrame基础操作文件名字段名说明air_data_points.csvMEMBER_NO会员卡号EXCHANGE_COUNT积分兑换次数EP_SUM总精英积分BP_SUM总基本积分AVG_BP_SUM每年平均基本积分POINTS_SUM总累计积分在PySpark中，创建DataFrame对象主要有以下几种方法：从结构化数据文件创建DataFrame：使用pyspark.sql.SparkSession.read模块，可以从多种格式的文件中读取数据并创建DataFrame。例如，从CSV、JSON、Parquet等文件格式中读取数据。从外部数据库创建DataFrame：通过pyspark.sql.SparkSession.read.format('jdbc')方法可以从关系型数据库表中读取数据并创建DataFrame。通过RDD创建DataFrame：如果已有一个pyspark.sql.SparkSession对象，并且有一个pyspark.rdd.RDD，可以通过调用toDF()方法将其转换为DataFrame。创建DataFrame对象pyspark.sql.SparkSession.read()方法是PySpark中用于从外部数据源读取数据并创建DataFrame的通用接口。它的完整调用格式为：

SparkSession.read(path,format=None,**options)path：字符串，表示要读取的数据的路径，可以是本地文件系统、HDFS、S3等。format：字符串，可选参数，表示数据的格式，如"csv"、"json"、"parquet"等。如果不指定，将根据文件扩展名自动推断格式。**options：字典，可选参数，表示特定于格式的选项，用于控制数据的解析方式。例如，对于CSV文件，可以使用option("header","true")来指定第一行为列名。创建DataFrame对象1.从结构化数据文件创建DataFrame此外，还有一些其他可选参数，可以通过options传递，schema，用于指定DataFrame的模式；loadOption，控制加载行为的选项，如是否忽略空白行、是否使用占位符替换缺失值等；header，指示输入文件是否包含列名作为第一行，默认为false；inferSchema，指示是否自动推断每列的数据类型。默认为false；comment，用于指定注释字符。在解析文件时，将以该字符开头的行为注释；charset，用于指定文件的字符集，默认为utf-8。更多参数详情，请参考官方文档。创建DataFrame对象现以读取存放在data目录下的某航空公司客户价值数据（3个CSV数据文件）为例进行说明。分别读取air_data_customer.csv、air_data_flight.csv和air_data_points.csv文件创建相应的DataFrame。spark.read.csv(‘../data/air_data_customer.csv’)也可写成其他语句形式，通过format(‘csv’)指定读取的文件格式为CSV，再通过load()方法指定文件的路径。创建DataFrame对象SparkSQL可以从MySQL、Oracle等外部数据库中创建DataFrame，使用这种方式创建DataFrame需要通过JDBC连接的方式访问数据库，并且对于特定的数据库需要使用特定的JDBC驱动JAR包。下面以MySQL数据库为例进行说明。准备工作：由于每个不同MySQL版本对应不同的JAR包，PySpark在使用MySQL时，需要将MySQL8.0.21的驱动JAR包mysql-connector-java-8.0.21.jar，复制到Anaconda创建的PySpark环境的jars子目录中（如:“D:\anaconda3\envs\pyspark\Lib\site-packages\pyspark\jars”），同时复制到Windows本地安装的JDK目录下的lib目录中，再重新启动进入PySpark。创建DataFrame对象2.从外部数据库创建DataFrame使用spark.read.format('jdbc')方法可以实现MySQL数据库的读取。在使用spark.read.format(‘jdbc’)方法时，可以通过链式调用option()方法来设置这些参数，以确保正确连接到MySQL数据库并高效地读取数据。总的来说，通过这些参数的合理配置，Spark能够以DataFrame的形式返回从MySQL数据库中读取的数据，进而可以利用SparkSQL的强大功能对数据进行分析和处理。创建DataFrame对象相关参数表格说明创建DataFrame对象参数名称参数的值含义urljdbc:mysql://localhost:3306/test数据库的连接地址drivercom.mysql.cj.jdbc.Driver数据库的JDBC驱动程序dbtablestudent要读取的表userroot数据库用户名password123456数据库用户密码将RDD数据转化为DataFrame有两种模式：反射模式、编程模式反射模式：若组成RDD[T]的每一个T对象内部具有公共且鲜明的字段结构，则可利用反射机制直接推断RDD模式。编程模式：若无法提前获知数据结构，则可使用编程模式定义一个Schema作为“表头”，将每条数据作为一个Row对象，并封装所有的Row对象作为“表中的记录”，将其转化为一个RDD，然后将“表头”和“表中的记录”拼接在一起，从而得到一个DataFrame，即带模式的RDD。需注意的是，RDD灵活性很高，并不是所有RDD都能转换为DataFrame，只有那些每个元素具有一定相似格式的RDD才可以转化为DataFrame。具体实现上，可使用toDF()方法和createDataFrame()方法创建DataFrame。创建DataFrame对象3.通过RDD创建DataFrame反射模式：通过RDD使用toDF()方法创建DataFrame若组成RDD[T]的每一个T对象内部具有公共且鲜明的字段结构，则可利用反射机制直接推断RDD模式。Listrdd中数据具有鲜明、统一的结构形式，结合其来源可推断每个元组由姓名、年龄和分数组成，于是可利用反射机制推断其模式，从而可将Listrdd转化为名为Listdf的DataFrame创建DataFrame对象使用createDataFrame()方法创建DataFrame使用createDataFrame()方法可利用反射机制直接从Python的列表、集合等类型数据中创建DataFrame，也可以从Pandas.DataFrame数据中创建DataFrame.创建DataFrame对象使用createDataFrame()方法创建DataFrame当无法事先获知数据结构时，可以使用createDataFrame()方法动态创建DataFrame。该方法需要指定RDD和Schema作为参数。创建DataFrame对象使用createDataFrame()方法创建DataFrame首先创建了一个SparkSession对象，然后定义了RDD数据和对应的Schema。接下来，通过调用createDataFrame()方法并传入RDD和Schema，成功创建了一个DataFrame对象。最后，使用show()方法展示DataFrame的内容。请注意，在实际应用中，需要根据具体的数据结构和需求来定义合适的Schema。创建DataFrame对象1DataFrame操作目录创建DataFrame对象2DataFrame输出操作3创建好DataFrame后，可利用DataFrame丰富、强大的API对DataFrame进行一系列数据处理操作，通过灵活组合使用这些DataFrameAPI实现业务需求。DataFrame提供DSL和SQL两种编程语法风格。二者在功能上并无区别，用户可根据习惯选择语法风格。DSL语法类似于RDD中的操作，允许开发者调用相关方法完成对DataFrame的操作。DSL语法风格符合面向对象编程的思想。SQL语法需要先将DataFrame注册成一个临时表，然后直接使用spark.sql()方法，该方法以SQL语句为参数，返回一个DataFrame对象。熟悉SQL语法的开发者，可以直接使用SQL语句进行操作。DataFrame操作DSL风格编程步骤：创建SparkSession：首先需要创建一个SparkSession对象，这是使用PySpark进行数据处理的入口点。加载数据：使用SparkSession的read方法加载数据，将数据封装到DataFrame中。例如，可以从CSV文件、JSON文件或数据库中读取数据。调用API函数：对DataFrame调用一系列的API函数进行处理和分析。这些函数类似于RDD中的操作，如select、filter、groupBy、agg等。执行操作：通过调用DataFrame的show、collect等方法来查看结果或执行其他操作。DataFrame操作SQL风格编程步骤：创建SparkSession：同样需要创建一个SparkSession对象。注册临时表：将DataFrame注册成一个临时表，这样可以使用SQL语句对其进行查询。使用createOrReplaceTempView方法来实现。编写SQL语句：编写SQL语句来查询临时表。SQL语句可以包含各种查询操作，如SELECT、INSERT、UPDATE、DELETE等。执行SQL：使用SparkSession的sql方法，将SQL语句作为参数传入，执行查询并返回结果的DataFrame。查看结果：可以通过DataFrame的show方法来查看查询结果。总的来说，DSL语法更符合面向对象编程的思想，而SQL语法则更适合熟悉SQL的开发者。在实际使用中，可以根据具体的应用场景和个人偏好选择合适的编程风格。DataFrame操作DataFrameAPI非常丰富，包括数据查看、数据处理、排序统计、连接及列操作等。1.数据查看API在PySpark中，DataFrame提供了一系列的API用于查看和操作数据。以下是一些常用的数据查看API。show()功能：展示DataFrame中的前N行数据，如果不指定参数，默认展示前20行。用法：df.show(n)，其中df是DataFrame对象，n是要展示的行数。DataFrame操作printSchema()功能：打印DataFrame的结构化模式，包括列名、数据类型和Nullable。用法：df.printSchema()。head()功能：返回DataFrame的前N行数据，通常用于快速查看数据的schema或前几行数据。用法：df.head(n)。DataFrame操作tail()功能：返回DataFrame的最后N行数据，常用于查看数据的最后一部分。用法：df.tail(n)。count()功能：返回DataFrame的行数。用法：df.count()。describe()功能：为DataFrame中的数值型列提供描述性统计信息，如均值、标准差、最小值、最大值等。用法：df.describe()。DataFrame操作summary()功能：为DataFrame中的所有列提供统计摘要，包括计数、均值、标准差、最小值、25%分位数、中位数、75%分位数和最大值。用法：df.summary()。dtypes功能：返回一个包含每列数据类型的DataFrame。用法：df.dtypes。DataFrame操作collect():collect()是一个action类型的API，它将DataFrame中的所有行以数组的形式返回到驱动程序中。这通常在结果集不是特别大时使用，因为它会将整个数据集加载到内存中。例如，df.collect()会返回DataFramedf中所有行的数组。take():take()同样是一个action类型的API，它从DataFrame中返回前N行数据。与collect()不同的是，take()不会将整个数据集加载到内存中，而是只获取部分数据。例如，df.take(5)会返回DataFramedf的前五行数据。需要注意的是，由于collect()和take()都是动作操作，它们会触发实际的数据计算并返回结果。在使用这些API时要谨慎，特别是在处理大型数据集时，因为可能会导致内存溢出或性能问题。而在进行数据分析时，通常会使用像show()、head()和tail()这样的转换操作来快速查看数据的概况。DataFrame操作在PySpark中，DataFrame提供了一系列的API用于数据处理，这些API可以帮助我们完成各种常见的数据操作。以下是一些常用的数据处理API及其详细说明：select(*cols)功能：从DataFrame中选取指定的一列或多列，返回一个新的DataFrame。用法：df.select(col1,col2,...)，其中df是DataFrame对象，col1,col2,...是要选取的列名。DataFrame操作2．数据处理APIselectExpr()

功能：选取列，并且可以对指定的列字段进行特殊处理，如聚合、算术运算等。用法：df.selectExpr("expr1","expr2",...)，其中expr1,expr2,...是要执行的表达式。where(conditionExpr)

功能：根据给定的条件筛选符合要求的行。用法：df.where(condition)，其中condition是一个表示筛选条件的表达式。DataFrame操作filter(conditionExpr)功能：与where()类似，也是根据给定的条件筛选符合要求的行。用法：df.filter(condition)。foreach(f)功能：将函数`f`应用于DataFrame的每一行。用法：df.foreach(f)，其中`f`是一个接受一行数据并对其进行操作的函数。DataFrame操作distinct()功能：去除DataFrame中的重复行，并返回去重后的DataFrame。用法：df.distinct()。alias(alias)功能：为DataFrame或其列设置别名，也可以将一个DataFrame复制为多个。用法：df.alias(alias)，其中alias是别名。DataFrame操作fillna()功能：对DataFrame中的空值使用指定的值进行填充。用法：df.fillna(value)，其中value是用于填充空值的值。dropna()功能：去除DataFrame中有缺失值的数据行，返回没有缺失值的新DataFrame。用法：df.dropna()。dropDuplicates()功能：删除DataFrame中的重复行，返回删除重复行后的新DataFrame。用法：df.dropDuplicates()。这些API提供了丰富的数据操作功能，使得DataFrame成为PySpark中一个非常强大的数据处理工具。DataFrame操作在PySpark中，DataFrame提供了一系列的API用于数据排序，这些API可以方便地对数据进行升序或降序的排序操作。以下是一些常用的数据排序API及其详细说明：orderBy()功能：根据指定的列对DataFrame进行排序，可以指定多个排序列和排序方向（升序或降序）。用法：df.orderBy(col1,col2,...)，其中df是DataFrame对象，col1,col2,...是要排序的列名。如果需要降序排序，可以使用desc()函数，如df.orderBy(col1.desc(),col2.asc())。DataFrame操作3．数据排序APIsort()功能：与orderBy()类似，也是用于对DataFrame进行排序，但它接受一个表示排序条件的字符串。用法：df.sort("col1","col2",...)，其中col1,col2,...是要排序的列名。同样，如果需要降序排序，可以使用desc()函数，如df.sort("col1desc","col2asc")。需要注意的是，排序操作通常会导致数据的重排，因此在处理大型数据集时可能会影响性能。在使用排序API时要谨慎，特别是在分布式环境中，因为排序操作可能会导致数据的移动和重新分布。DataFrame操作在PySpark中，DataFrame提供了一系列的API用于数据统计，这些API可以方便地对数据进行描述性统计和推断性统计分析。前面在数据查看API中介绍过的count()、describe()、summary()就是常用的统计API，以下也是一些常用的数据统计API。groupBy()功能：对数据进行分组，通常与聚合函数一起使用，如agg()、count()等，用于计算每个组的统计信息。用法：df.groupBy(col1,col2,...).agg(avg(col3),sum(col4),...)，其中col1,col2,...是分组列，col3,col4,...是需要进行聚合操作的列。DataFrame操作4．数据统计APIagg()功能：对DataFrame进行聚合操作，可以指定多个聚合函数，如avg()、sum()、min()、max()等。用法：df.agg(avg(col1),sum(col2),...)，其中col1,col2,...是要进行聚合操作的列。这些API使得DataFrame在PySpark中成为一个非常强大的数据统计工具，它们支持各种复杂的数据操作，包括但不限于数据筛选、转换、聚合和连接等。通过这些API，用户可以以编程的方式处理和分析大规模的数据集。DataFrame操作join()功能：join()方法用于将两个DataFrame根据共同的列或表达式进行连接。用法：df1.join(df2,on=joinExpression,how=joinType)参数说明：df1和df2是要连接的两个DataFrame。on参数是一个字符串（单列名）、列表（多列名）或表达式，指定了合并的基准。how参数指定了连接的类型，如"inner","outer","left_outer","right_outer","leftsemi"等。

DataFrame操作5．连接合并操作APImerge()功能：merge()方法提供了类似于数据库风格的连接，可以进行不同类型的合并操作。用法：DataFrame.merge(right,how,on,left_on,right_on,left_index,right_index,suffixes)参数说明：right是要与当前DataFrame进行合并的另一个DataFrame。how参数指定了合并的类型，如"inner","outer","left","right"等。on,left_on,right_on分别指定了合并的键。left_index和right_index参数允许使用DataFrame的索引作为合并的键。suffixes参数用于指定当键名称冲突时，添加到重叠列名后面的后缀。注意：使用join()、merge()API时，需要确保待连接的DataFrames有共同的列或表达式，以便正确执行合并操作。此外，不同的连接类型会影响最终结果的数据范围和结构，因此选择合适的连接类型对于数据处理至关重要。DataFrame操作以下是按照功能和使用方法对DataFrame的列操作API进行介绍。获取指定列：col()和apply()方法用于获取指定的列。col(name)：根据列名获取对应的列。apply(func,*args)：将一个函数应用于DataFrame的每一行，并返回一个新的DataFrame。删除列：drop(*cols)用于删除指定的列。参数colName指定删除列的名称。drop(colName1,colName2,...)删除多个指定名称的列。新增列：withColumn(colName,col)用于在DataFrame中添加新的列。其中colName是新列的名称，col是新列的值或表达式。DataFrame操作6．列操作API新增列：withColumn(colName,col)用于在DataFrame中添加新的列。其中colName是新列的名称，col是新列的值或表达式。改变列的顺序：select(*cols)可以改变DataFrame中列的顺序。select(col1,col2,...)按照指定的顺序选择列。重命名列：withColumnRenamed(existing,new)用于将现有名称为existing的列重命名为new。给列取别名：alias(*alias,**kwargs)方法可以为列设置别名。alias(colName,aliasName)为指定名称的列设置别名。alias(colName,