pyspark 入门：Spark SQL 练习的正确打开方式

很多朋友学 PySpark，一上来就啃 RDD 算子，什么 map、flatMap、reduceByKey，绕来绕去，最后写个 wordcount 就觉得自己入门了。但实际一回到真实业务，面对一堆 JSON、CSV、数据库表，你发现 RDD 那套东西写起来既啰嗦又不直观。我自己的经验是：如果你是新手，与其在 RDD 里挣扎，不如直接从 Spark SQL 切进去，用 DataFrame、用 SQL 语法去理解分布式计算，反而上手更快。

这篇文章就是围绕 pyspark 里的 Spark SQL 练习来写的。我会从环境准备、DataFrame 构建、SQL 操作、常见的坑，到最后一个完整练习案例，把实际干活时要用的东西都过一遍。适合刚开始学 PySpark、想做数据分析或数仓 ETL、又不想被概念绕晕的朋友。看完你就能在本地把 Spark 跑起来，用 DataFrame API 和 SQL 完成一套完整的数据分析流程。

1. 环境准备：先把 Spark 跑起来

1.1 本地安装与配置

说实话，本地装 Spark 没有想象中那么复杂，但版本匹配问题确实坑过不少人。我的建议是，先确定一个组合：Java 8 或 Java 11，Spark 3.3 或 3.4，Python 3.8 以上。这个组合我实测过多次，兼容性最省心。

步骤大概是这样的：

• 去官网下载 Spark 的预编译包，选带有 Hadoop 客户端的版本，比如spark-3.4.1-bin-hadoop3。
• 解压到一个不含空格的路径，比如D:\spark或/opt/spark。
• 配置环境变量SPARK_HOME，并把$SPARK_HOME/bin加入PATH。
• 安装 PySpark：pip install pyspark。注意，这里装的是 Python 端的库，它会自动寻找SPARK_HOME或者自己携带一个内置 Spark。

有一个特别容易踩的坑是 Python 解释器不一致。你在命令行输入python用的是某个版本，但 Spark 启动时可能去找系统默认的 Python，结果两边版本不一样，一执行就报Py4JJavaError或者Python worker failed to connect back。解决办法是显式设置PYSPARK_PYTHON和PYSPARK_DRIVER_PYTHON，让 Spark 知道你用的是哪个 Python。

在 Linux 或 Mac 上，可以在~/.bashrc里加：

export SPARK_HOME=/opt/spark export PATH=$SPARK_HOME/bin:$PATH export PYSPARK_PYTHON=python3 export PYSPARK_DRIVER_PYTHON=python3

Windows 上把路径改成你的实际安装位置，然后在系统环境变量里加同样的内容，改完记得重启终端。

1.2 两种运行方式怎么选

本地练习环境配好之后，你通常会遇到两种启动方式：一种是在 Jupyter Notebook 里跑，另一种是写脚本用spark-submit跑。这两种方式各有各的用处。

Jupyter 模式适合学习和调试，因为你可以一段一段地运行，随时看 DataFrame 的内容，查结果非常方便。在 Notebook 里需要加两行代码：

import findspark findspark.init()

如果没装 findspark，直接用pip install findspark就行。它的作用就是帮你在 Python 进程里自动设置 Spark 相关的环境变量，省得每次手动去配SPARK_HOME。

脚本模式适合批量跑任务，比如你写了一个analysis.py，然后用spark-submit analysis.py提交到集群或者本地运行。这种方式的好处是贴近生产环境，而且没有 Notebook 那种状态缓存的问题，每跑一次都是全新的会话。

我个人的建议是：新手先用 Jupyter 练操作，等逻辑成熟了，再封装成脚本用spark-submit跑一遍，感受一下两种模式的区别。很多人在 Notebook 里跑得好好的，一上脚本就各种报错，多半是因为环境变量没设置，并不是代码本身有问题。

2. 从零构建 DataFrame：Spark SQL 的第一块基石

2.1 从 RDD 转换

我知道很多人学 PySpark 入门时写过 RDD，所以这里先讲一下怎么把 RDD 转成 DataFrame。这是理解 Spark 数据模型的一个关键跳板。

from pyspark.sql import SparkSession spark = SparkSession.builder.appName("demo").getOrCreate() # 先造一个 RDD rdd = spark.sparkContext.parallelize([ ("Alice", 25, "engineer"), ("Bob", 30, "designer"), ("Cathy", 28, "manager") ]) # 方式一：toDF 指定列名 df1 = rdd.toDF(["name", "age", "job"]) df1.show()

toDF是最懒的方法，只要 RDD 里的元素是元组或者 Row 对象，就可以直接转。但这里有个隐藏问题：RDD 是没有 schema 的，里面的数据全是 Python 对象，Spark 在转换时要自己去推断类型。如果数据量很大，这个推断过程也会消耗时间。所以我更推荐下面这种方式，先手动定义 schema，再创建 DataFrame。

from pyspark.sql.types import StructType, StructField, StringType, IntegerType schema = StructType([ StructField("name", StringType(), True), StructField("age", IntegerType(), True), StructField("job", StringType(), True) ]) df2 = spark.createDataFrame(rdd, schema)

指定 schema 的好处很明显：类型可控、性能更好、不会出现“推断出个 bigint 还是 long 的差别”这种奇怪问题。后面读文件时同样建议手动指定 schema，尤其是生产环境。

2.2 从文件读取：CSV、JSON、Parquet

实际练习中，大部分数据都来自文件。PySpark 读取文件非常直接，但每个格式都有各自的细节。

先看 CSV，最常见，也最容易踩坑：

df = spark.read.csv( "data.csv", header=True, inferSchema=True, sep="," )

header=True表示第一行是表头，inferSchema=True表示让 Spark 自动推断字段类型。练习阶段这样写没问题，但如果你发现读进来某些列类型不对，比如日期变成了字符串，或者数字列有 null，就要考虑手动指定 schema 了。另一种情况是 CSV 里某些字段本身包含逗号，比如"张 三","工程师"这种带引号的格式，需要设置quote参数，否则会解析错位。

JSON 读取就更简单了：

df_json = spark.read.json("data.json")

但 JSON 有个天然的问题：嵌套结构。一行 JSON 里可能有个address字段，它本身是个字典。读进来之后，这个字段会变成 StructType，查询时要通过df["address.city"]这样点引用的方式去访问。新手很容易在这里被绕晕，建议先把数据拉平，或者用explode把嵌套数组拆开。

Parquet 是列式存储格式，也是 Spark 最推荐的存储格式。读取非常快，而且自带压缩和 schema 信息：

df_parquet = spark.read.parquet("data.parquet")

如果只是练习，不一定要特意生成 Parquet 文件，但你要知道：Spark 在读写 Parquet 时很多优化会自动生效，比如列裁剪和谓词下推。用 CSV 做练习体验不到这些特性，所以我建议后期一定要试试 Parquet。

2.3 从数据库读取：JDBC

这事练习阶段可能用不上，但既然做 pySpark，免不了有一天要直接从数据库读表。JDBC 读取要做几件事：首先确保数据库驱动 jar 已经让 Spark 能看到，其次连接串要写对。

df_db = spark.read.format("jdbc") \ .option("url", "jdbc:mysql://localhost:3306/test") \ .option("driver", "com.mysql.cj.jdbc.Driver") \ .option("user", "root") \ .option("password", "123456") \ .option("dbtable", "user_info") \ .load()

dbtable 那里可以写子查询，比如(select id, name from user_info where age > 18) tmp，这样 Spark 只拉符合条件的行过来，而不是把整张表全部读进来再做过滤，这是 JDBC 读取时一个非常实用的性能优化小技巧。注意，连接数据库前要把驱动 jar 放到 Spark 的 jars 目录里，或者在提交任务时用--jars指定，否则会报ClassNotFound错误。

2.4 Schema 的两种定义方式

前面提到过 schema 可以自动推断，也可以手动定义。这里展开说下，因为很多练习做着做着就卡在 schema 上。

自动推断适合第一次探索数据的时候，你想快速看看字段大概是什么类型。缺点是不稳定，数据一多容易推断出意想不到的类型，比如把字符串“12345”推断成 long。手动定义 schema 适合在明确字段结构的情况下使用，尤其是当字段很多、类型明确，或者来源是第三方数据接口时。

手动定义 schema 还有另一个好处：可以指定字段是否允许为 null。比如有的字段你觉得比较关键，不希望它为空，就在StructField里把第三参数设为False。这样 Spark 在写入或解析时就会直接报错，而不是把脏数据放过去，等于多了一层校验。

3. Spark SQL 核心操作实战

3.1 注册临时表与纯 SQL 查询

Spark SQL 最大的魅力就在于，你可以像操作关系型数据库那样写 SQL。先把 DataFrame 注册成临时表：

df.createOrReplaceTempView("user_info")

然后就可以用spark.sql()直接写查询了：

result = spark.sql(""" SELECT job, COUNT(*) AS cnt FROM user_info WHERE age >= 25 GROUP BY job ORDER BY cnt DESC """) result.show()

为什么有时候要用 SQL、有时候用 DataFrame API？我的经验是：SQL 适合表达复杂关联和聚合逻辑，读起来直白；DataFrame API 适合在代码里做条件分支、循环调用、动态拼接列名。两者可以随时互通，一张临时表既能用 SQL 查，也能用 API 继续操作，并不冲突。

有个小细节要提一下：createOrReplaceTempView创建的临时表仅在当前 SparkSession 内有效，一旦会话结束，表就没了，而且它不会进入 Hive 元数据。所以不要幻想临时表能跨会话复用。如果想让表持久化，可以用saveAsTable，它会把数据真正写到 Hive 仓库或 Spark 管理的存储中。

3.2 DataFrame API 常用操作

如果不想写 SQL，或者你的代码里需要动态处理列名，DataFrame API 就是主力。下面几个操作我几乎每个练习都会用到。

select用来选列，可以做列的重命名和表达式计算：

from pyspark.sql.functions import col, upper, year df.select( col("name"), upper(col("job")).alias("job_upper"), year(col("hire_date")).alias("hire_year") ).show()

注意这里的col("...")是构造一个 Column 对象，不是普通字符串。很多人刚学时写成df.select("name", "job")也行，但一旦涉及表达式、函数嵌套，就必须靠 Column 对象了。

filter也就是where条件过滤，支持字符串表达式和 Column 表达式两种写法：

# 字符串写法 df.filter("age >= 25 AND job = 'engineer'") # Column 写法 df.filter((col("age") >= 25) & (col("job") == "engineer"))

第二种写法的好处是不容易拼错引号，而且可以利用代码补全。注意多个条件用&连接而不是and，因为 Python 里的and不能重载 Column 对象的布尔运算，这是个很经典的报错点。

groupBy和agg是聚合操作的灵魂：

df.groupBy("job") \ .agg( count("*").alias("cnt"), avg("age").alias("avg_age"), max("age").alias("max_age") ) \ .show()

agg里可以放任意多个聚合函数，用alias给结果列起名字。这是最推荐的习惯，否则生成列名会是一串奇怪的默认名，后面引用的时候非常难受。

3.3 UDF：如何写自定义函数

内置函数总有覆盖不到的业务逻辑，这时候就要写 UDF。PySpark 的 UDF 其实很简单，但性能和序列化问题要注意。

举个例子，假设要根据年龄字段判断年龄段标签：

from pyspark.sql.functions import udf from pyspark.sql.types import StringType def age_group(age): if age is None: return "unknown" if age < 25: return "junior" elif age < 35: return "mid" else: return "senior" age_group_udf = udf(age_group, StringType()) df.withColumn("age_level", age_group_udf(col("age"))).show()

UDF 的核心逻辑是把 Python 函数序列化之后送到每个 Executor 的 Python worker 里去执行。既然是 Python 代码在跑，那就不会享受 Spark 原生算子的优化，所以性能会比内置函数差一些。我一般建议：能用内置函数解决的，就不要写 UDF。比如上面这个例子，用when函数也能实现，而且快得多：

from pyspark.sql.functions import when df.withColumn( "age_level", when(col("age").isNull(), "unknown") .when(col("age") < 25, "junior") .when(col("age") < 35, "mid") .otherwise("senior") ).show()

如果你确实要写 UDF，也建议启动 Pandas UDF，也就是pandas_udf。它的执行方式是按批次用 Pandas 来处理数据，性能比逐个 Python 函数调用高不少，API 却差不多，是实际生产里比较常用的折中方案。

3.4 用窗口函数实现分组 TopN

窗口函数是 Spark SQL 里特别实用的能力，尤其是分组取 TopN、累计求和、前后对比这些场景。很多新手不知道，用 groupBy 实现 TopN 会写得很绕，但窗口函数一条 SQL 就能搞定。

比如按部门分组，找每个部门薪水最高的前两个人：

from pyspark.sql.functions import row_number from pyspark.sql.window import Window window_spec = Window.partitionBy("dept").orderBy(col("salary").desc()) df.withColumn("rn", row_number().over(window_spec)) \ .filter(col("rn") <= 2) \ .drop("rn") \ .show()

这里的关键是Window.partitionBy(...).orderBy(...)定义了一个窗口范围：在每个dept分区内，按salary降序排列。row_number()会给每个分区内的行编上序号，然后在外面过滤rn <= 2就拿到了 Top 2。

这个写法要多练几次，因为窗口函数在数据分析里出现的频率超高。除了row_number，还有rank、dense_rank、lag、lead、sum开窗等等，它们的组合能解决很多“看起来很难”的业务问题。

4. 常见坑与性能排查实录

4.1 数据倾斜：任务卡在 99%

数据倾斜应该是分布式计算里最经典、也最让人头疼的问题。症状很典型：Spark 界面里某个 Stage 有十几个 Task，前面的 Task 几秒跑完，最后一个 Task 跑了几十分钟还没结束，甚至直接 OOM。

原因无非是某个 key 对应的数据量太大。比如按城市分组，可能某个城市的用户量是其他城市的几百倍，那么这个 key 所在的分区就会成为瓶颈。

处理数据倾斜的思路有不少，我实际用下来比较有效的有两个：

第一个是过滤掉异常大 key。如果这个超大 key 本身是脏数据或者不是业务关心的重点，直接过滤掉最简单，比如把 null、空串、默认值这些先剔除掉。

第二个是加随机前缀打散再聚合。做法是：先把一个聚合拆成两步，第一步给 key 加上随机前缀，让原本聚集在同一个分区的数据分散到多个分区去预聚合，第二步去掉前缀再做最终聚合。这里只演示第一个场景，第二个因为涉及两步聚合，逻辑相对复杂，练习阶段可以先了解思路，等遇到真实问题再回来实现。我个人遇到过最大的一次数据倾斜，就发生在一个用户行为表上，某个 app 渠道的埋点记录占了总量的一半，学会加随机前缀打散之后，整个任务从 40 分钟缩到了 6 分钟，效果立竿见影。

4.2 小文件问题：为什么越跑越慢

在分布式文件系统里，大量小文件是性能杀手。如果你的表每天写入几十个小文件，每个只有几十 KB，那么下游查询时，每个文件都要启动一个 Task 去读，白白浪费大量调度开销。

处理小文件最常用的手段是调整分区数和重分区。写入前可以用coalesce减少分区数，或者用repartition按某个列重新分区：

df.coalesce(1).write.mode("overwrite").parquet("output/") # 或者 df.repartition(col("dt")).write.mode("overwrite").parquet("output/dt=20250601")

这里coalesce(1)是把数据合并到一个分区再写出，适合小规模练习场景。生产环境一般不会这么干，因为单分区文件会太大，反而影响并发度。更通用的是根据数据量估算，比如目标每个文件 128 MB，就用文件总大小除以 128 MB 来算分区数。

但这里我要提醒一个容易误解的细节：repartition会触发全量 shuffle，很消耗资源。如果数据已经比较均匀，只是想把少量小文件合并一下，用coalesce更合适，因为它是在现有分区基础上合并，不会做完整的重洗牌。搞清楚这两个操作的区别，是优化写任务的第一步。

4.3 查询慢的隐藏原因：行式扫描 vs 列式裁剪

如果你发现自己明明写了select a, b from t，Spark 仍然把整张表的列都读进来了，那说明表可能是行式存储，或者没有利用到 Parquet 的列裁剪。用 CSV 和 JSON 做练习时，是不会有这种优化的，因为它们是行式存储格式，必须整行读出。

这也是我一直建议练习到后期要体验 Parquet 的原因。Parquet 是列式存储，Spark 在执行查询时，会先解析逻辑计划，只读取需要的列，来自动跳过无关列。如果你的表按照某个常用过滤字段做了分区，比如按日期dt分区，那查询时会直接跳过非目标分区的数据，这个叫分区裁剪。这些优化在 Spark 里都是自动发生的，但前提是你用对了格式和分区策略。

在实际练习中，你可以做一个简单对比：用一个 1 GB 左右的 CSV，分别以 CSV 格式和 Parquet 格式写出来，再去跑同一个聚合查询，看看耗时差异。实测下来 Parquet 通常能快好几倍，尤其是当你只查少数几列时，差距会非常明显。这样你就能直观感受到存储格式对查询性能的影响。

4.4 内存和序列化问题

PySpark 和纯 Spark 不一样的是，Python worker 和 JVM 之间需要频繁传递数据。如果你在一个 Python 函数里不小心引用了外部大对象，这个对象会被序列化并发送到每个 Executor，可能导致内存爆掉。

另一个常犯的错误是在 UDF 里直接引用 SparkSession。SparkSession 本身不能被序列化，一旦在 Executor 端尝试访问它，就会报出反序列化相关的错误。我在练习阶段踩过一次之后就记住了：UDF 里只放纯 Python 逻辑，凡是涉及 Spark 对象的东西都不要放进去。

如果遇到内存不足的报错，也别急着把内存参数往大了调。先看看代码里有没有不必要的collect()，这是把整个 DataFrame 拉回 Driver 的操作，数据量一大，Driver 端内存瞬间就能被打满。练习时用show()观察结果，生产环境用write落盘，只有明确知道数据量小、且需要回 Driver 做进一步处理时，才用collect()。

5. 完整练习：模拟用户行为数据分析

理论说再多，不如动手跑一个完整练习。这一节我造了一份模拟用户行为数据，包含注册信息、登录记录和订单流水，然后做一套常见的数据分析。你可以直接复制代码在本地跑，也可以改造成适合自己的数据集。

5.1 造数据

首先生成三份数据：用户表users、登录表logins、订单表orders。

from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType from pyspark.sql.functions import col, count, avg, sum, row_number, date_format from pyspark.sql.window import Window spark = SparkSession.builder.appName("spark_sql_practice").getOrCreate() # 用户数据 user_schema = StructType([ StructField("user_id", IntegerType(), False), StructField("name", StringType(), True), StructField("age", IntegerType(), True), StructField("city", StringType(), True) ]) users_data = [ (1, "Alice", 25, "Beijing"), (2, "Bob", 30, "Shanghai"), (3, "Cathy", 28, "Shenzhen"), (4, "David", 35, "Beijing"), (5, "Eve", 22, "Guangzhou") ] users = spark.createDataFrame(users_data, user_schema) # 登录数据 login_schema = StructType([ StructField("user_id", IntegerType(), False), StructField("login_date", StringType(), True), StructField("device", StringType(), True) ]) login_data = [ (1, "2025-05-01", "ios"), (1, "2025-05-02", "android"), (2, "2025-05-01", "web"), (3, "2025-05-01", "ios"), (3, "2025-05-02", "ios"), (4, "2025-05-01", "android"), (5, "2025-05-02", "web") ] logins = spark.createDataFrame(login_data, login_schema) # 订单数据 order_schema = StructType([ StructField("user_id", IntegerType(), False), StructField("order_id", StringType(), True), StructField("amount", DoubleType(), True), StructField("order_date", StringType(), True) ]) order_data = [ (1, "o1001", 99.9, "2025-05-01"), (1, "o1002", 159.0, "2025-05-02"), (2, "o1003", 299.0, "2025-05-01"), (4, "o1004", 450.0, "2025-05-02"), (4, "o1005", 23.5, "2025-05-03") ] orders = spark.createDataFrame(order_data, order_schema)

5.2 数据清洗与预处理

造完数据，第一步要处理一下字段格式和空值。这里我们把order_date转成标准日期格式，顺便去掉金额列里的 null 行，过滤掉年龄异常的记录：

clean_orders = orders.filter(col("amount").isNotNull()) \ .withColumn("order_dt", col("order_date").cast("date")) clean_users = users.filter((col("age") >= 0) & (col("age") <= 120))

注意cast("date")这种写法，是字符串转日期最方便的方式。如果转换失败，字段值会变成 null，所以转换完之后可以再查一遍空值数量。日期格式转换往往是练习里最琐碎但最重要的一步，后面按天分组、计算间隔都依赖它。

5.3 指标计算：用户维度、城市维度、TopN

先来两个最常用的聚合。第一个是每个用户的登录次数和总消费金额：

login_cnt = logins.groupBy("user_id").agg(count("login_date").alias("login_cnt")) order_sum = clean_orders.groupBy("user_id") \ .agg(sum("amount").alias("total_amount")) user_stats = users.join(login_cnt, "user_id", "left") \ .join(order_sum, "user_id", "left") \ .fillna(0) user_stats.show()

这里要注意fillna(0)只能填充数值列，如果你有字符串列为空，得用fillna("unknown", subset=["city"])。还有一点：两个 DataFrame 做 join 时，如果关联字段名字相同，直接在join里写"user_id"就行，Spark 会自动处理合并列；如果字段名不同，就要写成join(logins, users["user_id"] == logins["user_id"])，结果会保留两列。

第二个是按城市统计用户数、平均年龄、总订单金额：

city_stats = clean_users.join(order_sum, "user_id", "left") \ .groupBy("city") \ .agg( count("user_id").alias("user_cnt"), avg("age").alias("avg_age"), sum(col("total_amount")).alias("city_total_amount") ) \ .orderBy(col("city_total_amount").desc()) city_stats.show()

第三个是窗口函数练习：找出每个城市里订单金额最高的用户。

window_spec = Window.partitionBy("city").orderBy(col("total_amount").desc()) top_user_per_city = user_stats.withColumn("rn", row_number().over(window_spec)) \ .filter(col("rn") == 1) \ .select("city", "name", "total_amount") top_user_per_city.show()

这一步其实是把前面的知识串起来用了。你会发现一开始觉得窗口函数难懂，但只要配合完整的数据场景，多写几遍就能理解它的执行逻辑，无非就是“分组、排序、编号”这三件事。

5.4 结果输出与验证

练习的最后一步，把结果写到本地文件，并重新读回来验证一下：

user_stats.write.mode("overwrite").parquet("output/user_stats.parquet") city_stats.write.mode("overwrite").json("output/city_stats.json") # 读回来验证 check_df = spark.read.parquet("output/user_stats.parquet") check_df.show()

写文件时我习惯把输出目录放在项目下的output文件夹里，这样每次重跑之前清空一下目录，避免mode("overwrite")没生效时因为目录已存在而报错。这里有个细节：mode("overwrite")默认会覆盖整个目录，但如果你指定了format("json")，生成的文件并不是一个单个的.json文件，而是一个目录下多个 part 文件，这是正常的，不要一看到目录结构就以为自己写错了。

如果你希望结果是在一个文件里，可以用coalesce(1).write...，但前面说过，这只是练习场景或者结果集很小的情况下才推荐，因为数据量一大，单文件反而制约后续读取性能。验证完成后，可以在 Jupyter 里再跑几个验证 SQL，比如查一下城市排名结果是否符合预期，看看签到数与消费金额是否匹配，这些都能帮助发现前面步骤里可能埋下的 bug。

6. 关于学习和练习的一点心得

我这几年带过不少新人，也看过很多网上所谓的“入门教程”，最大的感受是：PySpark 入门容易，写出来能跑的代码也容易，但真正跑到大数据量上不崩、不慢、结果对，靠的是对 Spark 运行机制的理解和持续踩坑的积累。

如果你现在刚开始，我建议的学习路径是：先掌握 DataFrame 的构建和基础操作，再练 SQL 和 DataFrame API 的相互转换，然后用窗口函数解决实际问题，最后再去涉猎 RDD 底层、调优和集群部署。很多人一上来就从 RDD 开始学，结果学了一堆算子，遇到业务需求还是不知道怎么写 DataFrame，实在是绕了弯路。

最后分享一个小技巧：每次练习前，在代码开头打印一下 Spark 版本，然后强制设置合理的分区数。很多时候新手抱怨“Spark 跑得太慢”，其实不是 Spark 慢，而是默认的 200 个 shuffle 分区在本地环境里太夸张了，每个分区处理的数据太少，调度开销反而超过了计算时间。把spark.sql.shuffle.partitions调小一点，比如本地设成4，很多任务会明显变快。这个参数在集群上要重新评估，但在入门练习阶段，它帮你省下的时间能让你更专注于理解逻辑本身。