Spark电子书平台数据分析：从埋点日志到可视化报表

简介：面向大数据初学者的Spark电子书平台数据分析完整工程，围绕下载量统计、评分计算、阅读时段分析三类典型实时业务场景展开。项目以Scala语言实现，覆盖各标签总点击下载量实时统计、标签内下载Top10榜单、每本电子书评分动态更新，以及0:00-6:00、6:00-12:00、12:00-18:00、18:00-24:00四个时段在线人数监控等核心模块，可直接作为课程设计、毕业设计或Spark入门实战参考。压缩包共51个文件，主要包括36个scala源码、9个xml工程配置、2个docx项目文档、data样例数据与json结果文件，包体仅1.67MB，便于快速下载与本地运行。当前已有117人学习下载。附带u.data等测试数据及click_data.json输出，部署后即可观察各模块统计效果，项目文档还能帮助理解Spark实时分析的设计思路与实现要点。

1. 基于Spark的电子书平台数据分析设计与实现：为什么说这是最有性价比的大数据练手项目

做电子书平台的都知道，每天后台攒下的不只是订单，还有一长串用户行为：谁在搜书、谁点了收藏、谁把某本书读到第三章又退出。这些数据单个看没什么，堆到千万级之后，Excel 会崩，MySQL 单表聚合也开始变慢。基于 Spark 的电子书平台数据分析设计与实现，就是拿 Spark 把这种体量的日志和数据跑成运营能直接用的日报、画像和榜单。整套工程拆开看是三件事：设计数据模型、写 Spark 分析任务、把结果落到报表和可视化里。“源码+项目文档”的交付形式，决定了它不是讲概念的 PPT，而是一套能照着搭、能跑通、能讲清每个参数为什么这么配的完整样板。无论你是在为大数据岗位面试攒项目，还是想在离线数据分析方向沉淀一套可复用的技术模板，这个题目都值得投入。

2. 数据设计先行：Spark读取JSON的方式和电子书平台的表结构

数据分析项目最先翻车的往往不是代码，而是数据源没设计好。先花一整章把数据层讲透，后面所有分析才能站得住。这里按“日志怎么进”、“表怎么建”、“代码怎么摆”的顺序展开。

2.1 Spark中读取JSON：埋点日志的字段设计与read.json用法

电子书平台的前端埋点一般直接产生 JSON 日志，后端也可能把访问日志落成 JSON 文件。JSON 的好处是方便加字段，不需要提前建表；坏处是垃圾字段多、schema 不稳定。Spark 读取 JSON 没有黑科技，一行spark.read.json(path)就能读出一个 DataFrame，但生产环境不能这么裸奔。

来看一个完整的最小代码：

from pyspark.sql import SparkSession from pyspark.sql.types import ( StructType, StructField, StringType, IntegerType, TimestampType ) spark = SparkSession.builder \ .appName("ebook_log_read") \ .config("spark.sql.adaptive.enabled", "true") \ .getOrCreate() schema = StructType([ StructField("user_id", StringType(), True), StructField("book_id", StringType(), True), StructField("chapter_id", IntegerType(), True), StructField("event_time", TimestampType(), True), StructField("action", StringType(), True), StructField("read_duration", IntegerType(), True), StructField("device_type", StringType(), True), StructField("channel", StringType(), True) ]) logs_df = spark.read \ .option("ignoreCorruptFiles", "true") \ .schema(schema) \ .json("hdfs:///data/ebook/logs/dt=2025-03-15/*.json") logs_df.printSchema() logs_df.show(5, truncate=False)

这里两个关键点：手动给 schema，以及把spark.sql.adaptive.enabled打开。手动指定 schema 是为了防 Spark 自动推断时，把某个文件里的read_duration推断成字符串，导致后面的 sum 直接变成 null。ignoreCorruptFiles的作用是跳过损坏文件，而不是让整个任务红掉，这个参数在老集群上务必显式写。

参数方面，如果拿到的 JSON 每行是一个对象，用默认的multiLine=false就行；如果整个文件是一个大的 JSON 数组，需要加.option("multiLine", "true")。还有一个容易忽略的点：前端埋点常把页码、章节信息放在嵌套对象里，分析时用col("page_info.chapter_id")取子字段，或者用explode展开数组。字段名里可以有下划线，但不要用点号做字段分隔符，否则每次引用都要反引号，血泪经验。

2.2 表模型设计：用户维表、书籍维表、阅读行为事实表

数据分析的第一步是把业务关系摆正。电子书平台常见的建模方式是“两张维表 + 一张事实表”，也就是星型模型。维表变化慢，事实表每天会新增几百万行。

落地时，我一般用 Spark SQL 把表结构和字段约束写清楚，让任务每天用INSERT OVERWRITE刷新当日分区。

CREATE TABLE IF NOT EXISTS dwd_ebook_read_log ( user_id STRING COMMENT '用户ID', book_id STRING COMMENT '书籍ID', event_time TIMESTAMP COMMENT '事件时间', action STRING COMMENT 'read/collect/search/config', read_duration INT COMMENT '阅读时长(秒)', device_type STRING COMMENT 'pc/app/h5', channel STRING COMMENT '渠道' ) USING parquet PARTITIONED BY (dt STRING COMMENT '分区日期，如2025-03-15');

这张表后续所有分析都从它出数。事实表按天做分区，是因为电子书平台的分析基本是日报，一天一个分区既好管理，也方便重跑某一天。USING parquet是重点：Spark 读取 Parquet 比 JSON 快得多，Parquet 自带列裁剪，分析只需要user_id、book_id、read_duration三列时，不会把整个文件全部搬进内存。

2.3 最小项目目录结构：源码和项目文档该怎么摆

既然标题是“源码+项目文档”，代码组织就不能是单个 py 文件。见过太多项目把 ETL 和分析全写在一个 notebook 里，跑通一次容易，想重跑某一天时就麻了。我会按下面这个结构组织，边界清楚。

ebook-spark-analysis/ ├── src/ │ ├── main/python/ │ │ ├── parsing/ # 原始日志读取与schema定义 │ │ ├── etl/ # 清洗、去重、维度关联 │ │ ├── analysis/ # 日活、热榜、画像相关计算 │ │ └── export/ # 导出MySQL │ ├── main/resources/ │ │ └── application.conf # 数据库连接、路径、时间参数 ├── scripts/ │ └── submit.sh # spark-submit 跑批脚本 ├── docs/ │ ├── 01_需求文档.md # 指标口径、数据流图 │ └── 02_设计文档.md # 表模型、调度说明、参数说明 └── README.md

parsing 只负责“读出 DataFrame”，etl 只负责“清洗产出 dwd 表”，analysis 只负责“从 dwd 算指标”，export 只负责“写出去”。这样改清洗逻辑不用动分析代码，改输出方式也不用重跑整个链路。docs 里最该写清楚三件事：拿什么跑（spark-submit 参数）、跑哪一天的数据（时间参数）、结果在哪看（MySQL 表名和口径）。项目文档写得越像操作手册，这个项目的复现成本越低，面试讲的时候也越有底气。

3. Spark分析主流程：从数据清洗到核心指标计算

数据模型建好之后，进入 Spark 分析核心环节。这一章的代码可以直接抄进你的项目里，但一定要看我标出来的参数说明，因为同一个写法在不同集群上表现可能差很远。

3.1 数据清洗：过滤脏数据、去重、统一时间口径

电子书平台日志最常见的脏数据有三种：user_id 为空、book_id 不存在、event_time 不在当天分区里。清洗规则不会太复杂，但要把所有条件落在一个函数里，避免散落各处。

from pyspark.sql import functions as F def transform_read_log(raw_df): return (raw_df .filter(F.col("user_id").isNotNull() & F.col("book_id").isNotNull() & F.col("event_time").isNotNull()) .filter(F.col("event_time").between( "2025-03-15 00:00:00", "2025-03-15 23:59:59")) .dropDuplicates(["user_id", "book_id", "chapter_id", "event_time"]) .withColumn("dt", F.to_date(F.col("event_time"))))

逻辑说明：第一个 filter 去掉空主键，第二个 filter 把时间窗固定到分析当天。dropDuplicates 按“同一用户对同一本书的同一章节在同一个时间点”去重。注意这里带上了 chapter_id，目的是保留用户在同一个章节的多次连续阅读记录，而不是把整本书的阅读记录合并成一条。withColumn生成 dt 字段，后续所有分区裁剪都用它。

参数说明：event_time.between对 timestamp 类型是闭区间，所以结束时间要写当天的 23:59:59，不要写第二天 00:00:00，否则会包进第二天零点整的数据。这个边界我至少踩过两次，每次都是日报比业务方手算多几条记录，最后定位到这里。

3.2 核心指标：UV、PV、阅读时长聚合

覆盖大部分需求的核心指标是日活不重复用户数（UV）、页面浏览数（PV）和总阅读时长。Spark 里countDistinct是精确去重，在数据量大时很贵；如果是看板类允许偏差的场景，用approx_count_distinct可以省不少 shuffle。

daily_stats = (clean_df .groupBy("dt") .agg( F.countDistinct("user_id").alias("uv"), F.count("*").alias("pv"), F.sum("read_duration").alias("total_read_seconds"), F.avg("read_duration").alias("avg_read_seconds") ) .orderBy("dt")) daily_stats.show()

这里要关注的是 Spark 在计算countDistinct时的内存行为。数据量到几亿行，distinct 会触发大 shuffle，如果遇到某本爆款书占了 80% 日志，一个 executor 的堆内存很容易被打满。这时候先考虑approx_count_distinct，或者把倾斜键单独拎出来算完再合并。另外sum("read_duration")之前要确认字段是整数类型，否则结果全是 null，这种问题最常出现在 JSON 推断出字符串列之后。

书籍热度的计算会增加一个维度，按书聚合：

book_rank = (clean_df .groupBy("book_id", "dt") .agg( F.countDistinct("user_id").alias("readers"), F.sum("read_duration").alias("total_read_seconds") ) .filter(F.col("readers") > 10) .orderBy(F.col("total_read_seconds").desc()) .limit(100)) book_rank.write.parquet("hdfs:///result/ebook/book_rank/dt=2025-03-15")

filter("readers > 10")是常见套路，过滤掉只有一两个人读的长尾书，让榜单聚焦在真实内容上。limit(100) 只在最终结果取 100 本，不会影响前面的聚合。写到 Parquet 按日期分区，后续可视化直接读目录，不用再算一遍。

3.3 用户画像标签：行为权重打分

用户画像听起来玄学，落到代码里其实就是给行为打权重再聚合。电子书平台的标签不需要实时推荐那样的复杂度，离线打分足够支撑运营。规则是：阅读行为按分钟数计分，收藏一次计 2 分，搜索一次计 1 分，按用户汇总出对每个书籍类别的兴趣分。

category_score = (clean_df .join(F.broadcast(dim_book), "book_id", "left") .groupBy("user_id", "category") .agg( F.sum( F.when(F.col("action") == "read", F.col("read_duration") / 60) .when(F.col("action") == "collect", 2.0) .when(F.col("action") == "search", 1.0) .otherwise(0.0) ).alias("category_score") ) .filter(F.col("category").isNotNull())) category_score.show(20)

关键在F.when / .otherwise链式写法，把一个阅读事件按动作类型换算成可累加的分数。read_duration以秒为单位除以 60 变分钟，解决“读了一篇长文”和“点开 3 秒就退出”的行为分值差异。用F.broadcast(dim_book)是因为书籍维表通常只有几十万行，广播到每个 executor 后 join 不会产生 shuffle，这是常用的性能优化。

3.4 一个Spark数据分析案例的扩展：用ALS做简单的书单推荐

如果想让项目文档里多一个“个性化推荐”模块，最务实的是用 Spark MLlib 的 ALS 做协同过滤，而不是自己写余弦相似度。ALS 对“用户—书籍—交互分数”矩阵天然友好，封装也完整。

from pyspark.ml import Pipeline from pyspark.ml.feature import StringIndexer from pyspark.ml.recommendation import ALS user_idx = StringIndexer(inputCol="user_id", outputCol="user_idx") book_idx = StringIndexer(inputCol="book_id", outputCol="book_idx") pipeline = Pipeline(stages=[user_idx, book_idx]) interaction_df = (clean_df .groupBy("user_id", "book_id") .agg(F.sum(F.col("read_duration") / 60).alias("interact_score")) .filter(F.col("interact_score") > 1)) data = pipeline.fit(interaction_df).transform(interaction_df) als = ALS( maxIter=10, regParam=0.05, userCol="user_idx", itemCol="book_idx", ratingCol="interact_score", coldStartStrategy="drop" ) model = als.fit(data) recs = model.recommendForAllUsers(5)

interact_score过滤掉阅读不足 1 分钟的弱交互，减少噪音。coldStartStrategy="drop"必须设置，否则对新用户或新书会输出 null，写表时直接报错。ALS 训练结束后recommendForAllUsers(5)返回每个用户 Top5，但开销不小，实际跑批通常只对近 7 天活跃用户做推荐，而不是全量用户。这个模块放进 spark数据分析案例里是很自然的扩展，不复杂，但能把项目从“只会 groupBy”提升到“能跑机器学习”的层次。

4. 结果落地：MySQL导出、Python可视化和定时跑批

分析结果不落库，项目文档写得再漂亮也白搭。这一章处理“数据从哪里出去”的问题，也是把 Spark 作业变成稳定服务的关键一步。

4.1 用JDBC写回MySQL：写入模式与批次参数

把 Spark 聚合结果写回 MySQL，最直接的是 DataFrame 的jdbc方法。下面是每日报表的落库代码：

import os daily_stats.write \ .mode("overwrite") \ .option("truncate", "true") \ .jdbc(url="jdbc:mysql://localhost:3306/ebook_analysis", table="daily_report", properties={ "user": "root", "password": os.environ.get("MYSQL_PASSWORD", ""), "driver": "com.mysql.cj.jdbc.Driver", "rewriteBatchedStatements": "true", "batchsize": "5000" })

逻辑说明：mode("overwrite")配合truncate=true，在 MySQL 端先清空目标表再写入，适合每日全量结果表。这样重跑任务时 MySQL 不会因为残留旧数据产生重复记录。daily_stats只有几千行，写库很快；如果是要写几亿行明细，不建议这个做法，overwrite会很慢，应该先落临时目录再 LOAD DATA。

参数说明：rewriteBatchedStatements=true是 MySQL 批量写入的经典参数，能把多个 insert 合并成一次网络往返。batchsize默认 1000，调到 5000 通常更快，但设过大会让 MySQL 端内存压力上升。密码不要写死在代码里，从环境变量读，最后由 spark-submit 注入。

4.2 用Python做数据分析与可视化，画趋势图

Spark 适合算数，不适合画图。最终呈现在看板上，一般是把结果表拉回 Python，用 pandas 加 matplotlib 处理。不要想着 Spark 自带图表，也别在集群节点上跑带 GUI 的 Python，那会把人折腾死。

import os import pandas as pd import matplotlib.pyplot as plt from sqlalchemy import create_engine password = os.environ.get("MYSQL_PASSWORD", "") engine = create_engine(f"mysql+pymysql://root:{password}@localhost:3306/ebook_analysis") report = pd.read_sql( "SELECT dt, uv, pv, total_read_seconds FROM daily_report ORDER BY dt", engine) report["dt"] = pd.to_datetime(report["dt"]) fig, ax = plt.subplots(figsize=(12, 5)) ax.plot(report["dt"], report["uv"], label="UV", marker="o") ax.plot(report["dt"], report["pv"], label="PV", marker="o") ax.set_title("电子书平台日活与浏览量趋势") ax.legend() plt.show()

这个做法的前提是 daily_report 数据量不大，一次性read_sql没问题。真正的生产系统里，看板后端通常只查最近 30 天数据，或者用接口返回给前端图表。本地做数据分析与可视化时，这一套足够。

补充一个容易踩的细节：Spark 写进 MySQL 的dt字段是字符串类型，Python 侧要主动转成 datetime，否则 matplotlib 会把字符串当类别处理，横轴顺序会乱。另外total_read_seconds单位是秒，展示时最好在 Python 里除以 3600 变成小时，不要直接在数据库里改字段语义。

4.3 Spark搭建教程里必带的submit脚本与定时调度

集群上的 Spark 作业不能像本地开发那样在 IDE 里点运行，都要走spark-submit。我平时把跑批命令写成脚本放在scripts/submit.sh，新人照着目录文档就能跑。

#!/usr/bin/env bash set -euo pipefail DT=${1:-$(date -d "yesterday" +%F)} APP_JAR="/opt/ebook-analysis/ebook-analysis.jar" spark-submit \ --master yarn \ --deploy-mode cluster \ --name "ebook_analysis_daily_${DT}" \ --class com.example.EbookAnalysis \ --driver-memory 2g \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 4 \ --files /opt/ebook-analysis/conf/application.conf#application.conf \ "${APP_JAR}" \ --dt "${DT}"

脚本里的每个参数都有含义：--master yarn表示走 YARN 集群调度；--deploy-mode cluster让 Driver 也在集群里运行，本地客户端提交后即可退出；--driver-memory要结合collect的量来设，如果只在 Driver 上做了少量小结果展示，2g 足够；executor-memory4g、executor-cores2 是很多离线任务的起步配置。真正调优时去看 Spark UI 的 Executors 页，如果 GC 时间占比高就加内存，如果 CPU 利用率低就减少 core 数或增加并发。

定时调度日常用 cron 就能满足：

30 2 * * * /opt/ebook-analysis/scripts/submit.sh $(date -d "yesterday" +%F) >> /var/log/ebook-spark.log 2>&1

Spark 任务跑 30 分钟还是 3 个小时，和资源队列、数据量强相关。cron 只是“触发”，不能兜底失败重跑，所以脚本里set -euo pipefail很有必要：任何一步失败就非零退出，日志才会有明确报错。更工程化一些可以接 DolphinScheduler 或 Airflow，但小项目用 cron 加日志足够。

5. 避坑清单：基于Spark的电子书数据分析项目的5个高频问题

下面这些坑都是我实际跑这类项目撞过的，每一条按“现象、原因、解决”三个关键点写，可以直接当排查手册用。

5.1 现象：读入大量JSON小文件时任务卡在FileScan阶段

现象：Spark 读取一天的 JSON 日志，日志量只有 2GB，但文件数有上万个，任务在 FileScan 阶段就花了 40 分钟。原因：每个小文件都会产生一个 partition，Spark 调度上万个 partition 的开销远大于数据处理本身。解决：先跑一次小文件合并任务，把 JSON 统一转成 Parquet 并按日期分区：

spark.read.schema(schema).json("hdfs:///data/ebook/logs/2025-03-15/*.json") \ .coalesce(50) \ .write.mode("overwrite") \ .partitionBy("dt") \ .parquet("hdfs:///data/ebook/dwd/read_log")

这里coalesce(50)把结果控制在 50 个文件左右，不是越大越好。partitionBy("dt")让后续查询可以自动做分区裁剪，只读当天目录。如果原始文件已经是整天的目录，优化后第二次跑同样指标通常快 5 倍以上。

5.2 现象：Executor lost / OOM，容器被YARN杀掉

现象：作业在聚合阶段突然失败，日志显示某个 executor 的 container killed，或者java.lang.OutOfMemoryError。原因：多数是数据倾斜加内存配置不对。比如某个热门小说占当天 80% 的阅读日志，按 book_id groupBy 时，处理那本书的任务全部落在同一个 executor 上，堆被撑爆。解决：先在 Spark UI 的 Stage 详情里找到 skew 的 task，再把倾斜的 key 加盐：

salt = F.col("book_id").cast("string") + "_" + F.rand().cast("int") skewed_clean = clean_df.withColumn("salt_book_id", salt) part_agg = (skewed_clean .groupBy("salt_book_id") .agg(F.countDistinct("user_id").alias("readers"), F.sum("read_duration").alias("total_read_seconds"))) # 第二遍聚合：去掉盐，还原成真实book_id final = (part_agg .withColumn("book_id", F.regexp_replace("salt_book_id", "_\d+$", "")) .groupBy("book_id") .agg(F.sum("readers").alias("readers"), F.sum("total_read_seconds").alias("total_read_seconds")))

加盐后同一本书的数据会分散到 10 个 key 上，每个 reducer 只处理 1/10 数据，但还要再做一次去盐聚合把结果还原。更稳的做法是把spark.sql.shuffle.partitions调大，并且给 executor 的spark.memory.overhead留足余量，比如--conf spark.memory.overhead=1g。Spark 内存配置有时候就是玄学，看到 OOM 先加 overhead，再调 shuffle 分区。

5.3 现象：本地模式跑通，Spark集群搭建好后却跑不出结果

现象：代码在--master local[2]下跑得飞快，提交到集群就报错，错误千奇百怪，比如找不到 MySQL 驱动、读不到 application.conf。原因：本地模式自动把代码和配置放在同一个 JVM，而 YARN 集群模式里代码在远端容器执行，driver 和 executor 拿不到之前的 classpath 和文件。解决：提交时显式带上资源文件：

spark-submit ... \ --files /opt/ebook-analysis/conf/application.conf#application.conf \ --conf spark.executor.extraJavaOptions=-Dconfig.file=application.conf \ --jars /opt/ebook-analysis/mysql-connector-j-8.0.33.jar

务必用--files而不是让代码里写死本地路径。另一点：本地直接读 MySQL 没问题，集群上如果 executor 访问 MySQL 被网络策略挡住，也会出现本地成功集群失败，这个要先确认安全组和连通性。

5.4 现象：读取JSON时schema推断结果每天都不一样

现象：某天跑出的read_duration变成了 string，导致 sum 报异常；再过一天又变成 long，一个作业要改三回代码。原因：spark.read.json在小文件数量多时，流程是“先采样再推断”，采样到的文件里恰好缺字段或类型变化，就会把 schema 推断错。解决：用 2.1 节的方式手动指定 schema，并且打开spark.sql.json.ignoreCorruptFiles。schema 必须手工维护，这也是项目文档里最该写明的第一原则。

5.5 现象：在Driver端collect()后内存瞬间爆掉

现象：分析完成后用collect()拉结果到本地，大结果集直接 OOM，或者集群 client 节点卡死。原因：collect()会把所有分区的数据全部传到 Driver，Driver 堆内存只有--driver-memory那么大，结果集一旦超过它就死路一条。解决：只预览数据就用take(10)或show()；需要全量结果就去写 MySQL 或 HDFS，不要 collect。如果确需在 Driver 做二次处理，也要先repartition减小数据量，并且调大--driver-memory：

sample_preview = clean_df.repartition(1).take(10)

repartition(1)不推荐用于大结果，会做一次全量 shuffle，但用于 take 预览可接受。日常开发时，我会在 Spark UI 的 SQL 页看 Physical Plan，凡是看到 Exchange 节点有大量分区，再确认 shuffle 增量是否合理。

6. 从跑通到跑稳：用记录数核对来验证数据管道

很多项目跑通一次之后，第二次换了时间参数，结果数字对不上，最后发现是清洗时把 null 全部丢掉了。这个问题的根源不是代码 bug，而是缺少验证步骤。我养成的习惯是每个作业末尾都输出“血缘指标”：读进来多少条、过滤后剩多少条、聚合后多少行，然后写进一张data_quality_log表，下次跑批直接对比。

raw_count = raw_df.count() clean_count = clean_df.count() detail_count = daily_stats.count() quality_df = spark.createDataFrame([ ("2025-03-15", "raw_log", raw_count, "日志原始条数"), ("2025-03-15", "clean_log", clean_count, "清洗后条数"), ("2025-03-15", "daily_report", detail_count, "日报结果行数") ], ["dt", "metric", "cnt", "note"]) quality_df.write.jdbc( url="jdbc:mysql://localhost:3306/ebook_analysis", table="data_quality_log", mode="append", properties={ "user": "root", "password": os.environ.get("MYSQL_PASSWORD", ""), "driver": "com.mysql.cj.jdbc.Driver" })

这三个数字能说明很多问题：如果clean_count比昨日同口径少了 50%，大概率是数据源路径写错或清洗条件过严；如果daily_report行数为 0，那问题多半出在 dt 参数传递上。不要小看这几行代码，它相当于给黑盒式的 Spark 作业装了一个仪表盘。

在项目文档里，我会把验证步骤写成三档：第一档是“能跑通”，即 spark-submit 提交后 exit code 为 0；第二档是“能对上”，即关键指标和业务方日报历史值偏差在可接受范围；第三档是“能自动恢复”，即失败重跑不会产生重复数据，靠的是写 MySQL 时的写入模式。做完这三档，基于 Spark 的电子书平台数据分析设计与实现才算真正交付，而不仅仅是交一段代码。

最后说一个我自己的习惯：每次提交跑批任务之前，先用本地小数据集把--dt参数写死跑一遍，确认今天的日期、路径和分区都存在，再上集群。这个动作帮我省了至少十次因为日期写错导致的无效跑批。希望帮到你。

本文还有配套的精品资源，点击获取