Article

数据计算 Spark

更新于：2026-07-13

一、开始

1.1 spark-shell

操作	说明	注意事项
启动 spark-shell	在命令行输入 `spark-shell` 即可启动交互式环境。	确保 Spark 已正确安装并配置环境变量。
引入依赖	`spark-shell --jars /path/myjar1.jar,/path/myjar2.jar`	多个 jar 包使用逗号分隔。
自动加载 `sc`	`sc` 即 `org.apache.spark.SparkContext`，spark-shell 启动后自动创建。	无需手动初始化，可直接使用。
自动加载 `spark`	`spark` 即 `org.apache.spark.sql.SparkSession`，spark-shell 启动后自动创建。	无需手动初始化，可直接使用。
设置日志级别	`spark.sparkContext.setLogLevel("ERROR")`	减少日志输出，便于观察结果。可选级别：ERROR、WARN、INFO、DEBUG。

1.2 IntelliJ IDEA 配置

1.2.1 Maven 依赖配置（pom.xml）

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.finance.data</groupId>
    <artifactId>analyst</artifactId>
    <version>1.0</version>

    <properties>
        <spark.version>2.4.7</spark.version>
        <scala.version>2.11.12</scala.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-avro_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <!--
            encrypt 为三方库，远程服务器没有对应 jar 包，需要以坐标的方式安装：
            mvn install:install-file -DgroupId=com.fibodt.encrypt -DartifactId=rule -Dversion=1.0 -Dpackaging=jar -Dfile=rule-1.0.jar
            提交至集群时，需要使用 jars 参数提供 jar 包
            -->
            <groupId>com.fibodt.encrypt</groupId>
            <artifactId>rule</artifactId>
            <version>1.0</version>
        </dependency>
    </dependencies>

    <build>
        <finalName>analyst</finalName>
        <plugins>
            <!-- 编译 Scala 代码 -->
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <version>2.15.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                        <configuration>
                            <sourceDir>src</sourceDir>
                            <includes>
                                <include>**/*.scala</include>
                            </includes>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <!-- 编译 Java 代码 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.3.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <!-- 在 package 阶段，所有依赖均被剔除 -->
                                    <artifact>*:*</artifact>
                                    <!--
                                    当调用胖包报错"Invalid signature file digest for Manifest main attributes"时，
                                    需要将后缀为 SF、DSA、RSA 的文件剔除
                                    -->
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <!-- 指定主函数 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>3.2.0</version>
                <configuration>
                    <archive>
                        <manifest>
                            <addClasspath>true</addClasspath>
                            <mainClass>com.finance.data.main.task</mainClass>
                        </manifest>
                    </archive>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

1.2.2 定义 spark 和 sc

操作	代码	注意事项
定义 `spark`	`val spark = SparkSession.builder().appName("Word Count").enableHiveSupport().getOrCreate()`	master 可以不在代码中指定，使用 `spark-submit` 的 `--master` 参数会覆盖代码中的配置。
定义 `sc`	`sc = spark.sparkContext()`	通过 SparkSession 获取 SparkContext。

1.2.3 延迟计算

概念	说明	注意事项
lazy 修饰符	使用 `lazy` 修饰成员变量，避免初始化时间过长。	当 Spark 代码以 object 单例对象的成员变量/方法形式存在时，如果包含 action 算子且计算开销较大，强烈建议加 `lazy`，防止初始化时间过长。

二、快速入门

2.1 文件读取

val textFile = sc.textFile("/Users/lijp/IdeaProjects/sc/src/main/java/sc/汽车品牌.csv")
// textFile 为 RDD 类型，具有 List 的很多相似操作（map、foreach、filter 等）

2.2 基本操作

操作	说明
`map`	对 RDD 中每行进行处理。
`flatMap`	对 RDD 中每行进行展开处理。
`collect`	将结果转换为 Array 类型。
`cache`	将 RDD 和 Dataset 保存在内存中，被 session 持有。

2.3 spark-submit 提交

2.3.1 单机模式

spark-submit --class package包路径 --master local[4] jar包路径

2.3.2 集群模式（YARN）

提示：如果集群使用者过多可能会因为尝试 20 次申请资源导致失败，可增加配置 --conf spark.port.maxRetries=1000。注意：金融集群使用 spark2-submit，其余参数维持不变。

spark-submit --class package.path.to.classfile \
--queue yarn.dig3 \
--master yarn \
--driver-memory 10g \
--num-executors 60 \
--executor-memory 10g \
--executor-cores 2 \
--total-executor-cores 200 \
--conf spark.port.maxRetries=1000 \
--conf spark.dynamicAllocation.enabled=true \
--conf spark.dynamicAllocation.maxExecutors=100 \
--conf spark.dynamicAllocation.minExecutors=10 \
--conf spark.shuffle.service.enabled=true \
--conf spark.default.parallelism=800 \
--conf spark.sql.shuffle.partitions=800 \
--conf spark.driver.maxResultSize=2g \
--jars ./tools/RuleEncrypt-1.0-SNAPSHOT.jar \
./tools/original-rule-analyst.jar 20220101 20220101 "" click

2.3.3 spark-submit 参数说明

选项	说明	注意事项
`--class`	指定应用程序的入口类。	必填。
`--master`	指定 Spark 集群的主节点 URL。例如 `--master yarn` 表示在 YARN 集群上运行。	必填。
`--deploy-mode`	指定部署模式：`client`（提交节点启动 Driver）或 `cluster`（集群中启动 Driver）。	根据场景选择。
`--driver-memory`	设置 Driver 使用的内存资源，一般为 executor-memory 的 20%~25%。	影响 Driver 端数据处理能力。
`--driver-cores`	设置 Driver 使用的核心数。	Spark 3.1.1 及以上版本支持。
`--executor-memory`	设置每个 Executor 的内存资源。	`num-executors * executor-memory` 不能超过 Max Resources 中的 memory 数量。
`--executor-cores`	设置每个 Executor 的核心数。	`num-executors * executor-cores` 不能超过 Max Resources 中的 vCores 数量。
`--num-executors`	设置 Executor 的数量，等价于 `--conf spark.executor.instances=10`。	—
`--conf`	设置 Spark 配置参数，如 `--conf spark.shuffle.compress=true`。	支持大量配置项，详见下文。
`--files`	指定要分发到集群的文件，多个文件使用逗号分隔。文件以原始格式存储到任务的工作目录。	如果加载的是本地文件需配置；HDFS 文件无需配置，Spark 会自动分发。
`--archives`	指定要分发到集群的压缩文件。任务执行时自动解压。	主要用于传递 Python 项目第三方依赖库。
`--jars`	指定要添加到类路径的 JAR 文件，等价于 `--conf spark.jars=...`。多个文件使用逗号分隔。	支持本地文件、HDFS 路径、file:// 路径三种形式。
`--packages`	在运行时自动下载并添加 Maven 依赖包，如 `--packages org.example:tqdm:1.0.0`。	指定 Maven 坐标信息，自动下载依赖。
`--py-files`	指定要添加到 Python 环境的 `.zip`、`.egg` 或 `.py` 文件。	用于传递 Python 项目，支持目录/压缩包/压缩包 HDFS 路径。
`--name`	指定应用程序的名称。	—
`--queue`	指定应用程序提交到的队列。	同一用户可有多个资源队列。
`--verbose`	在输出中显示详细信息。	调试时建议开启。

2.3.4 `--files` 使用说明

通过 --files 可将一个或多个普通文件复制到每个任务的工作目录中，通过相对路径访问。

操作	代码	注意事项
启动 spark-shell（传递文件）	`spark-shell --files hdfs:///user/lijp/temp/3_shanghai_20231013.txt`	文件分发到各节点家目录下。
读取文件（正确方式）	`val df = spark.read.csv("temp/3_shanghai_20231013.txt")`	使用相对路径 `temp/...` 读取。
读取文件（错误方式）	`val df = spark.read.csv("3_shanghai_20231013.txt")`	报错 `Path does not exist`，因为文件仅分发到了 `/user/lijp` 下。

2.3.5 `--archives` 使用说明

操作	代码	注意事项
启动 spark-shell（传递压缩文件）	`spark-shell --archives hdfs:///user/lijp/temp/test.zip`	文件传递过程中为压缩格式，任务执行期间自动解压缩。
读取文件	`val df = spark.read.csv("/temp/archive/test/rule_d_84_20231017.txt")`	使用相对路径 `/temp/...` 读取解压后的文件。

2.3.6 `--py-files` 使用说明

场景	命令/代码	注意事项
导入 Python 项目文件夹	`--py-files your_project`	—
导入 Python 项目压缩包	`--py-files your_project.zip`	启动任务时分发到各节点，运行时自动解压。
导入 HDFS 上文件夹	`--py-files hdfs:///user/lijianping/tools/finance_analysis`	—
导入 HDFS 上压缩包	`--py-files hdfs:///user/lijianping/tools/finance_analysis.zip`	—
项目内部 import（相对路径）	`from ..share.sparks import spark`	Python 项目内部必须使用相对路径导入。
pyspark 命令行 import（绝对路径）	`from finance_analysis.process.shb_return import shb_return_process`	命令行中使用绝对路径导入。

注意：子模块中的 __init__.py 需要添加以下代码确保导入正常：
import sys
import os
current_dir = os.path.dirname(os.path.abspath(__file__))
sys.path.append(current_dir)

2.3.7 常见 `--conf` 选项

选项	样例	说明
`spark.app.name`	`--conf spark.app.name=test`	配置应用名称，低版本 Spark 可能无法生效。
`spark.driver.extraLibraryPath`	`--conf spark.driver.extraLibraryPath=/opt/cloudera/parcels/CDH-5.15.0-1.cdh5.15.0p0.21/lib/hadoop/lib/native`	指定 Driver 进程加载本地库（.so/.dll/.dylib）。
`spark.dynamicAllocation.enabled`	`--conf spark.dynamicAllocation.enabled=true`	启用动态资源分配，自动分配和回收执行器资源。
`spark.dynamicAllocation.minExecutors`	`--conf spark.dynamicAllocation.minExecutors=1`	最小执行器数量。
`spark.dynamicAllocation.maxExecutors`	`--conf spark.dynamicAllocation.maxExecutors=10`	最大执行器数量。
`spark.dynamicAllocation.executorIdleTimeout`	`--conf spark.dynamicAllocation.executorIdleTimeout=60`	执行器闲置超时时间（默认 60 秒），超时后释放。
`spark.dynamicAllocation.schedulerBacklogTimeout`	`--conf spark.dynamicAllocation.schedulerBacklogTimeout=1`	等待调度器作业请求的超时时间（默认 1 秒）。
`spark.dynamicAllocation.shuffleTracking.enabled`	`--conf spark.dynamicAllocation.shuffleTracking.enabled=true`	根据 shuffle 操作动态分配资源。
`spark.dynamicAllocation.executorAllocationRatio`	`--conf spark.dynamicAllocation.executorAllocationRatio=1.0`	可用内核与执行器数量比例，默认 1.0。
`spark.eventLog.enabled`	`--conf spark.eventLog.enabled=true`	保存事件日志。
`spark.eventLog.dir`	`--conf spark.eventLog.dir=/path/to/logs`	事件日志 HDFS 存放路径。
`spark.jars`	`--conf spark.jars=tools/original-analyst.jar,RuleEncrypt-1.0.jar`	引入依赖 jar 包路径。
`spark.lineage.enabled`	`--conf spark.lineage.enabled=true`	启用血统跟踪功能，用于故障时重新计算数据。
`spark.lineage.log.dir`	`--conf spark.lineage.log.dir=/var/log/spark/lineage`	血统日志存放路径。
`spark.repl.class.uri`	`--conf spark.repl.class.uri=http://example.com/my-lib.jar`	为 spark-shell 加载外部类/库。
`spark.scheduler.mode`	`--conf spark.scheduler.mode=FIFO`	任务调度模式：FIFO（先进先出）、FAIR（公平）、LOAD_BALANCING（负载均衡）。
`spark.shuffle.service.enabled`	`--conf spark.shuffle.service.enabled=true`	启用 Shuffle 服务，减轻执行器内存压力。
`spark.driver.maxResultSize`	`--conf spark.driver.maxResultSize=2g`	Driver 收集结果数据最大允许大小，默认 1g。
`spark.default.parallelism`	`--conf spark.default.parallelism=800`	RDD 操作并行度。
`spark.sql.shuffle.partitions`	`--conf spark.sql.shuffle.partitions=800`	Spark SQL shuffle 分区数，默认 200。大数据集聚合失败时适当调大。
`spark.sql.hive.verifyPartitionPath`	`--conf spark.sql.hive.verifyPartitionPath=true`	验证 Hive 分区路径。
`spark.memory.overhead`	`--conf spark.memory.overhead=5g`	应用程序整体内存超额分配，建议为 `spark.executor.memory` 的 10%~20%。
`spark.executor.memoryOverhead`	`--conf spark.executor.memoryOverhead=5g`	每个 Executor 的内存超额分配，建议为 `spark.executor.memory` 的 10%~15%。
`spark.executor.instances`	`--conf spark.executor.instances=100`	执行器实例数量，等价于 `--num-executors 100`。
`spark.executor.memory`	`--conf spark.executor.memory=4g`	每个执行器内存大小，等价于 `--executor-memory 4g`。
`spark.driver.memory`	`--conf spark.driver.memory=10g`	Driver 内存大小，等价于 `--driver-memory 10g`。
`spark.executor.cores`	`--conf spark.executor.cores=4`	每个执行器 CPU 核心数，等价于 `--executor-cores 4`。
`spark.memory.offHeap.enabled`	`--conf spark.memory.offHeap.enabled=true`	启用堆外内存。
`spark.memory.offHeap.size`	`--conf spark.memory.offHeap.size=10g`	堆外内存大小。
`spark.storage.memoryFraction`	`--conf spark.storage.memoryFraction=0.6`	可用于存储缓存的内存比例，默认 0.6（60%）。
`spark.shuffle.memoryFraction`	`--conf spark.shuffle.memoryFraction=0.2`	可用于 Shuffle 操作的内存比例，默认 0.2（20%）。
`spark.executor.heartbeatInterval`	`--conf spark.executor.heartbeatInterval=100s`	执行器与 Driver 心跳间隔。
`spark.network.timeout`	`--conf spark.network.timeout=1000s`	网络超时时间，防止因网络延迟导致超时错误。
`spark.sql.sources.partitionOverwriteMode`	`--conf spark.sql.sources.partitionOverwriteMode=dynamic`	分区数据覆盖模式，`dynamic` 允许动态覆盖。
`spark.yarn.queue`	`--conf spark.yarn.queue=queue_name`	YARN 队列名称。
`spark.yarn.am.memory`	—	ApplicationMaster 内存大小。
`spark.yarn.am.cores`	—	ApplicationMaster CPU 核心数。
`spark.yarn.am.memoryOverhead`	—	ApplicationMaster 内存开销。
`spark.yarn.driver.memoryOverhead`	—	Driver 进程内存开销。
`spark.yarn.executor.memory`	—	YARN 执行器内存大小。
`spark.yarn.executor.cores`	—	YARN 执行器 CPU 核心数。
`spark.yarn.executor.instances`	—	YARN 执行器实例数量。
`spark.yarn.appMasterEnv.PYSPARK_PYTHON`	`--conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./project_env/env/bin/python3`	YARN ApplicationMaster 中使用的 Python 解释器路径。
`spark.yarn.appMasterEnv.PYSPARK_DRIVER_PYTHON`	`--conf spark.yarn.appMasterEnv.PYSPARK_DRIVER_PYTHON=./project_env/env/bin/python3`	YARN Driver 程序中使用的 Python 解释器路径。

2.4 队列资源说明

2.4.1 队列资源指标

指标	说明
Used Resources	当前已使用的资源，包括内存和虚拟内核（vCores）。
Demand Resources	当前正在请求的资源。
AM Used Resources	ApplicationMaster 已使用的资源。
AM Max Resources	ApplicationMaster 可使用的最大资源。
Num Active Applications	当前正在运行的应用程序数量。
Num Pending Applications	当前等待资源的应用程序数量。
Min Resources	队列允许的最小资源限制。
Max Resources	队列允许的最大资源限制。
Reserved Resources	当前被保留的资源。
Steady Fair Share	队列的稳定公平份额。
Instantaneous Fair Share	队列的瞬时公平份额。
Preemptable	是否允许抢占式资源调度。

2.4.2 参数设定指导

以实际集群资源为例：

每个执行器最大内存：25600M（25G）
driver-memory 最大：40G
driver-cores 最大：10
executor-memory 最大：1200G（总计）
executor-cores 最大：500（总计）

步骤	操作	说明
1. 确定 executor-memory	`25G / 11 * 10 = 22.72G`	`--executor-memory` 实际需要的内存为其值的 1.1 倍（含 overhead），不能超过单执行器上限 25G。
2. 确定 num-executors	`1200G / 22.72G = 52.82`	Max Resources 下 memory 总量除以单执行器内存，得到执行器数量上限。
3. 确定 executor-cores	`500 / 53 = 9.43`	Max Resources 下 vCores 总量除以执行器数量，得到每个执行器核心数上限。
4. 微调参数	executor-memory: 22 → 20，num-executors: 53 → 50，executor-cores: 9 → 10	参数取整，重新计算。

2.4.3 参数依赖关系

参数	依赖内存	依赖并行
`--num-executors`	大文件	多文件、多分区
`--executor-memory`	大文件、聚合、排序、复杂中间结果、广播变量	—
`--executor-cores`	—	多文件、多分区、计算密集、多线程计算
`--driver-memory`	本地数据聚合、广播变量、小数据集计算、三方库操作	—
`--driver-cores`	—	—

最终提交命令示例：

spark-submit --class package.path.to.classfile \
--queue yarn.dig3 \
--master yarn \
--driver-memory 10g \
--num-executors 60 \
--executor-memory 10g \
--executor-cores 2 \
--conf spark.port.maxRetries=1000 \
--jars ./tools/RuleEncrypt-1.0-SNAPSHOT.jar \
./tools/original-rule-analyst.jar 20220101 20220101 "" click

三、RDD 编程指引

3.1 创建 RDD

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data, 5)
// distData 类型为 ParallelCollectionRDD，分片数为 5

RDD 可以看作是 Spark 分布式环境下的 List。

3.2 读取文件

注意事项	说明
本地文件	需要在所有节点上可以被访问到。
目录/通配符/压缩包	`sc.textFile("/my/directory")`、`sc.textFile("/my/directory/.txt")`、`sc.textFile("/my/directory/.gz")`
控制返回文件数	使用 `SparkContext.wholeTextFiles` 控制返回文件个数。
SequenceFile	`SparkContext.sequenceFile[Int, String]`
HadoopRDD	`SparkContext.hadoopRDD`
ObjectFile	`SparkContext.objectFile`

3.3 RDD 操作

3.3.1 注意事项

注意点	说明
元素格式	RDD 内部元素标准形式应为 `(key, value)`（长度为 2），不符合此形式时部分接口不支持。
parallelize 与 df.rdd	`sc.parallelize` 生成的 RDD 元素为元组，支持所有接口；`df.rdd` 生成的 RDD 元素为 Row，只支持部分接口。

样例数据：

val rdd = sc.parallelize(1 to 9, 3)
val df1 = List(("one", 1, 11), ("two", 2, 22), ("three", 3, 33), ("four", 4, 44), ("five", 5, 55)).toDF("cnt", "int", "ints")
val df2 = List(("one", 5, 55), ("two", 4, 44), ("three", 3, 33), ("four", 2, 22), ("five", 1, 11)).toDF("cnt", "int", "ints")
val rdd1 = df1.rdd.map(row => (row.getAs[String]("cnt"), row.getAs[Int]("int"), row.getAs[Int]("ints")))
val rdd2 = df2.rdd.map(row => (row.getAs[String]("cnt"), row.getAs[Int]("int"), row.getAs[Int]("ints")))

3.3.2 转换（Transform）

转换操作用懒加载，生成新的 RDD，不立即执行。

3.3.2.1 map

返回一个新的分布式数据集，通过将源的每个元素传递给函数 func 形成。

rdd.map(line => line + 1).collect()

3.3.2.2 mapPartitions

与 map 相似，但分别在 RDD 的每个分区（块）上运行，func 对整个分区进行处理（Iterator<T> => Iterator<U>）。

// 将每个分区视为一个整体，返回各分区最大值
rdd.mapPartitions(iter => List(iter.max).toIterator).collect()

3.3.2.3 mapPartitionsWithIndex

与 mapPartitions 相似，额外提供表示分区索引的整数值（(Int, Iterator<T>) => Iterator<U>）。

// 带入当前分区索引值
rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).collect()

3.3.2.4 mapValues

在 (K, V) 对的数据集上调用，仅对 V 进行操作。

rdd.map(line => line -> line.toString).mapValues(line => line + "_").collect()

3.3.2.5 【早期版本】mapWith

引入两个函数：一个处理分片索引，另一个接受当前分片数据和处理完的索引，生成新分片数据。

// 第一个函数 index => index * 2 处理分片索引
// 第二个函数接受当前分片数据和处理后的 index
rdd.mapWith(index => index * 2)((iter, index) => iter.map(i => index -> i)).collect()

3.3.2.6 flatMap

与 map 相似，但每个输入项可以映射到 0 个或多个输出项（func 返回 Seq 而非单个项）。

// map 步骤：根据传入函数生成新 Iterator
// flat 步骤：将每个 Iterator 展开合并为一个扁平的 Iterator
rdd.flatMap(line => List(line).toTraversable).collect()
rdd1.flatMap(line => line.productIterator.toTraversable).collect()

3.3.2.7 filter

返回 func 返回 true 的元素形成的新数据集。

rdd.filter(line => line % 2 == 0).collect()

3.3.2.8 sample

参数	说明
`withReplacement`	`true` 表示有放回抽样，原数据集大小不变；`false` 表示无放回抽样，数据集减少。
`fraction`	抽样比例。
`seed`	随机数种子（Long 型，如 `12345L`）。

rdd.sample(true, .5).collect()

3.3.2.9 union

返回源数据集中元素的并集。合并前会检查类型一致性。

// 类型不一致时报错（如 RDD[(Int, Int)] 不能和 RDD[Int] 合并）
rdd.union(rdd).collect()

3.3.2.10 intersection

返回源数据集中元素的交集，不针对键，针对元素。需类型一致。

rdd.intersection(rdd.filter(line => line % 2 == 0)).collect()

3.3.2.11 distinct

返回源数据集中不同元素的新数据集。

rdd.union(rdd).distinct

3.3.2.12 groupBy

根据函数返回值对 RDD 重新分组。

// 单列元素
rdd.groupBy(line => line % 2).collect()
// (K, V) 对
rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).groupBy(line => line._1 % 2).collect()
// (K, V1, V2) 按 K 分组
rdd1.union(rdd2).groupBy(line => line._1).collect()

3.3.2.13 groupByKey

在 (K, V) 对数据集上调用，返回 (K, Iterable<V>)。仅按键分组，不聚合值。

注意事项	说明
仅按键分组	无法根据值进行聚合。
性能建议	如果要对每个键执行聚合（求和/平均值），使用 `reduceByKey` 或 `aggregateByKey` 性能更好。
并行度	默认取决于父 RDD 分区数，可传 `numPartitions` 参数。

rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).groupByKey(2).collect()
// reduceByKey 等效效果
rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).groupByKey(2).map(line => (line._1, line._2.reduce((x, y) => x + y))).collect()

3.3.2.14 reduce（Action）

将 RDD 中元素两两传递给函数，产生新值继续处理，直至只有一个值。是 Action 操作，无需 collect。

// 一元 RDD：reduce 函数类型 (Int, Int) => Int
rdd.reduce((x, y) => x + y)
// 键值对 RDD：函数类型 ((Int, Int), (Int, Int)) => (Int, Int)
rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).reduce((x, y) => (x._1 + y._1, x._2 + y._2))

3.3.2.15 reduceByKey

在 (K, V) 对数据集上调用，返回 (K, V)，每个键的值使用 reduce 函数汇总。

rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).reduceByKey((x, y) => x + y).collect()
// 配置并行度为 3
rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).reduceByKey((x, y) => x + y, 3).collect()

3.3.2.16 aggregate（Action）

Action 操作，无需 collect 即可返回结果。初值同时参与分区内和分区间聚合。

// 简单求和
rdd.aggregate(0)((x, y) => x + y, (x, y) => x + y)
// 初值为 1，观察聚合过程：初值参与每个组内聚合和组间聚合的第一步计算
rdd.aggregate(1)((x,y) => {println("组内聚合",x,y);x + y}, (x,y) => {println("组间聚合:", x, y);x + y})

3.3.2.17 aggregateByKey

在 (K, V) 对数据集上调用，返回 (K, U)。允许与输入值类型不同的聚合值类型。与 aggregate 不同的是，初值不参与组间聚合。

// 简单求和
rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).aggregateByKey(0)((x, y) => x + y, (x, y) => x + y).collect()
// 求均值
rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i))
  .map(line => (line._1, (line._2, 1)))
  .aggregateByKey((0,0))((x, y) => (x._1 + y._1, x._2 + y._2), (x, y) => (x._1 + y._1, x._2 + y._2))
  .map(line => (line._1, line._2._1 / line._2._2)).collect
// 求和（初值类型与 V 类型不一致时）
rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i))
  .map(line => (line._1, (line._2, 1)))
  .aggregateByKey(0)((x, y) => x + y._1, (x, y) => x + y)
  .collect

3.3.2.18 sortByKey

在 K 实现 Ordered 的 (K, V) 对数据集上调用，按键排序。

rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).sortByKey(false).collect() // 降序

3.3.2.19 join

在 (K, V) 和 (K, W) 类型上调用，返回 (K, (V, W))。

注意事项	说明
类型要求	join 在两个 `(K, V)` 类型 RDD 数据集上发生，非 `(K, V)` 类型需使用 `intersection`。
类型确认	join 前建议确认元素类型，防止 Any 类型导致报错。

val rdd1 = rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).filter(line => line._1 % 2 == 0)
val rdd2 = rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i)).filter(line => line._1 % 3 == 0)
rdd1.join(rdd2).collect() // inner join
rdd1.leftOuterJoin(rdd2).collect() // left join
rdd1.rightOuterJoin(rdd2).collect() // right join
rdd1.fullOuterJoin(rdd2).collect() // full outer join

3.3.2.20 cogroup

在 (K, V) 和 (K, W) 类型上调用，返回 (K, (Iterable<V>, Iterable<W>))。相当于先在各自 RDD 上 groupByKey，再 fullOuterJoin。

rdd1.cogroup(rdd2).collect()

3.3.2.21 cartesian

返回两个数据集所有元素对的笛卡尔积 (T, U)。

3.3.2.22 pipe

通过 shell 命令管道传输 RDD 的每个分区。

3.3.2.23 coalesce

将 RDD 分区数减少到 numPartitions。适用于筛选大型数据集后高效运行操作。

3.3.2.24 repartition

随机重排 RDD 中的数据以创建更多或更少的分区，在整个分区之间保持平衡。

rdd.repartition(1) // 合并为一个分区

3.3.2.25 repartitionAndSortWithinPartitions

根据分区程序对 RDD 重新分区，并在每个分区内按键排序。比先 repartition 再 sort 更高效。

3.3.3 行动（Action）

汇总所有结果返回 Driver。

操作	说明
`reduce`	使用函数聚合数据集元素。函数需可交换和可结合。
`collect`	将数据集所有元素作为数组返回 Driver。仅适用于数据量较小的情况。
`count`	返回数据集中元素的数量。
`first`	返回数据集第一个元素（等同于 `take(1)`）。
`take`	返回数据集前 n 个元素的数组。
`takeSample`	返回包含 num 个随机样本元素的数组。
`takeOrdered`	使用自然顺序或自定义比较器返回前 n 个元素。
`saveAsTextFile`	将元素以文本文件形式写入指定目录。每个元素调用 toString 转为一行文本。
`saveAsSequenceFile`	将元素作为 Hadoop SequenceFile 写入。适用于实现 Writable 接口的键值对 RDD。
`saveAsObjectFile`	使用 Java 序列化格式写入，可用 `SparkContext.objectFile()` 加载。
`countByKey`	仅在 `(K, V)` 类型 RDD 上可用，返回每个键计数的哈希图。
`foreach`	在数据集每个元素上执行副作用操作（如更新累加器、与外部系统交互）。

3.3.4 缓存

操作	说明
`persist`	可按参数指定不同缓存级别：`MEMORY_ONLY`、`MEMORY_AND_DISK`、`MEMORY_ONLY_SER`、`MEMORY_AND_DISK_SER`、`DISK_ONLY`。
`cache`	默认缓存级别 `MEMORY_ONLY`。
级别选择	`MEMORY_ONLY` > `MEMORY_ONLY_SER` > `MEMORY_AND_DISK`。
`unpersist`	释放缓存。

3.3.5 打印部分记录

操作	说明	注意事项
`collect`	将全部记录汇总到一台机器上。	可能耗尽内存。
`take`	获取部分记录。	推荐用于预览数据。

3.3.6 共享变量

3.3.6.1 广播变量

在所有节点上创建一个只读变量。DataFrame 和变量可使用 broadcast 广播，但 RDD 不可以。

val broadcastVar = sc.broadcast(Array(1, 2, 3))
val broadcastDF = functions.broadcast(df)

3.3.6.2 累加器

创建累加器：

val accum = sc.longAccumulator("My Accumulator")
sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
println(accum.value)

自定义累加器：

// 继承 AccumulatorV2
class VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] {
  private val myVector: MyVector = MyVector.createZeroVector
  def reset(): Unit = { myVector.reset() }
  def add(v: MyVector): Unit = { myVector.add(v) }
}
// 创建并注册
val myVectorAcc = new VectorAccumulatorV2
sc.register(myVectorAcc, "MyVectorAcc1")

注意：累加器依赖 Action 操作触发执行，在惰性求值无 Action 触发时不会被更新。

3.4 数据类型

类型	说明
`Array`	可转换为 List：`Array.toList`。
`WrappedArray`	可转换为 Array：`WrappedArray.array.toArray`。

3.5 使用举例

创建样例 RDD：

val rdd = sc.parallelize(1 to 9, 3)
val kvrdd = rdd.mapPartitionsWithIndex((index, iter) => iter.map(i => index -> i))

3.5.1 构造 (K, V) 类型（用于 join）

kvrdd.map(line => (line._1, line._2._2))

3.5.2 SELECT 操作

kvrdd.map(line => (line._1, line._2))

3.5.3 WHERE 操作

kvrdd.filter(line => line._1 % 2 == 0)

3.5.4 聚合操作

groupByKey 方式

// 求最大值
kvrdd.groupByKey().map(line => (line._1, line._2.max))
// 求最小值
kvrdd.groupByKey().map(line => (line._1, line._2.min))
// 求均值
kvrdd.groupByKey().map(line => (line._1, line._2.sum / line._2.size))
// 求和
kvrdd.groupByKey().map(line => (line._1, line._2.sum))

reduceByKey 方式

// 求最大值（两两比较）
kvrdd.reduceByKey((x, y) => List(x, y).max)
// 求最小值（两两比较）
kvrdd.reduceByKey((x, y) => List(x, y).min)
// 求均值（需匹配常数列用于计数）
kvrdd.map(line => (line._1, (line._2, 1)))
  .reduceByKey((x,y) => (x._1 + y._1, x._2 + y._2))
  .map(line => (line._1, line._2._1 / line._2._2)).collect
// 求和（两两求和）
kvrdd.reduceByKey((x, y) => x + y)

aggregateByKey 方式

// 求最大值
kvrdd.aggregateByKey(0)((x,y) => List(x, y).max, (x, y) => List(x,y).max).collect()
// 求最小值
kvrdd.aggregateByKey(0)((x, y) => List(x, y).min, (x, y) => List(x, y).min).collect()
// 求均值
kvrdd.map(line => (line._1, (line._2, 1)))
  .aggregateByKey((0, 0))((x, y) => (x._1 + y._1, x._2 + y._2), (x, y) => (x._1 + y._1, x._2 + y._2))
  .map(line => (line._1, line._2._1 / line._2._2)).collect
// 求和
kvrdd.aggregateByKey(0)((x, y) => x + y, (x, y) => x + y).collect

join 操作

val kvrdd1 = rdd.map(line => (line, line)).filter(line => line._1 % 2 == 0)
val kvrdd2 = rdd.map(line => (line, line)).filter(line => line._1 % 3 == 0)
// inner join
kvrdd1.join(kvrdd2).collect()
// left join
kvrdd1.leftOuterJoin(kvrdd2).collect()
// right join
kvrdd1.rightOuterJoin(kvrdd2).collect()
// full outer join
kvrdd1.fullOuterJoin(kvrdd2).collect()

3.6 RDD API 计算矩阵乘法

计算过程：

将矩阵 A 转置，使列索引 j 变为键，以便与矩阵 B 的行索引 j 对齐
按索引 j 进行 join
在每个组合内生成 (i, k) 坐标，计算临时结果 v * w
按 (i, k) 坐标 groupBy，对临时结果求和
结果解读：(i, k) 为矩阵坐标，sum(v * w) 为该坐标上的值

val m = sc.parallelize(Seq(
  (0, 0, 1.0), (0, 1, 2.0),
  (1, 0, 3.0), (1, 1, 4.0)
)) // 定义矩阵 M

val n = sc.parallelize(Seq(
  (0, 0, 5.0), (0, 1, 6.0),
  (1, 0, 7.0), (1, 1, 8.0)
)) // 定义矩阵 N

val res = m.map(m => (m._2, (m._1, m._3))) // M 转换为 ((j, i), v)
  .join(
    n.map(n => (n._1, (n._2, n._3))) // N 转换为 ((j, k), w)
  )
  .map {
    case (j, ((i, v), (k, w))) => ((i, k), v * w)
  }
  .reduceByKey(_ + _) // 求和

res.foreach(println) // 输出结果

四、SparkSQL / DataSets / DataFrames

4.1 读取文件

读取方式	代码示例	说明
JSON	`val df = spark.read.json("examples/src/main/resources/people.json")`	读取 JSON 文件。
CSV	`val df = spark.read.csv("file:///D:/java_workspace/fun_test.csv")`	读取 CSV 文件。
JDBC	`spark.read.jdbc`	通过 JDBC 读取数据库表。
ORC	`spark.read.orc`	读取 ORC 格式文件。
Parquet	`spark.read.parquet`	读取 Parquet 格式文件。
TextFile	`spark.read.textFile`	读取文本文件。
format + load	`val df = spark.read.format("com.databricks.spark.avro").load("/raw_data/operator/8/dpi_result_fp/p_biz={e_17,e_20}")`	通过 `format` 指定格式后加载。
option 设置参数	`spark.read.option("header","true").csv("/Users/lijp/IdeaProjects/testspark/src/main/scala/sample.txt")`	使用 `option` 设置读取参数。
schema 设置列类型	见下方代码示例	当自动推导类型失败时，需指定 schema 进行解析。

使用 schema 指定列类型（解决自动推导类型失败问题）：

// 假设 csv 文件中列的值较为复杂，可能导致自动推导类型失败，需要指定 schema 进行解析
// 假设 csv 文件内容：G2_12345678910|LOAN,SMS,mobile,1,1,1,1,1,0,0,-1,-1
val schema = StructType(StructField("uid", StringType) :: StructField("info", StringType) :: Nil)
val df = spark.read.schema(schema).csv("/user/lijianping/xykd_sample")

4.2 显示数据

操作	代码	说明
展示数据	`df.show()`	默认显示前 20 行。
显示 Schema	`df.printSchema()`	打印 DataFrame 的 Schema 结构。
显示不截断	`df.show(false)`	防止长列名尾部出现省略号。

4.3 选择数据

4.3.1 仅选择列

df.select("name").show()

4.3.2 选择并计算

// Spark 2.4.5
df.select("name","age+1").show()
df.select("name",s"age+${value}").show()
// Spark 2.4.0
df.select($"name", $"age"+1).show()

4.3.3 过滤

// Spark 2.4.5
df.filter("age>18").show()
df.filter(s"age>${value}").show()
// Spark 2.4.0
df.filter($"age">18).show()

对整数类型过滤

方式	示例	说明
逻辑运算符	`df.filter($"num"===2)`、`df.filter($"num">2)`、`df.filter($"num"<2)`	`===`、`>`、`<`、`!==`
字符串表达式	`df.filter("num=2")`、`df.filter("num>2")`、`df.filter("num<2")`	—
传递参数过滤	`val ind:Int=2; df.filter($"num"===ind)`	支持 `===`、`>`、`<`

对字符串过滤

方式	示例	说明
equalTo	`df.filter($"id".equalTo("a"))`	—
传递参数过滤	`val str = "a"; df.filter($"id".equalTo(str))`	—
默认字段名	`df.filter($"_1".equalTo("a"))`	当 DataFrame 没有字段名时可用 `_1`、`_2` 等默认字段名。
多条件判断	`df.filter($"num"===2 && $"id".equalTo("a"))`	`&&`（并）、`\|\|`（或）

4.3.4 NA 处理

操作	代码	说明
丢弃含 NA 的行	`df.na.drop()`	—
丢弃少于两个值的行	`df.na.drop(thresh = 2)`	保留至少 2 个非空值的行。
丢弃全为 NA 的行	`df.na.drop(how='all')`	—
丢弃有一个 NA 的行	`df.na.drop(how="any")`	—
针对指定列丢弃	`df.na.drop(cols=Array('Sales'))`	仅在 Sales 列上判断 NA。
用 0 填充	`df.na.fill(0)`	—
针对指定列填充	`df.na.fill(value="no label", cols=Array("label"))`	—
对不同列填充不同值	`df.na.fill(valueMap=Map("stringCol" -> "test", "intCol" -> 100))`	填充值类型需与 schema 要求一致。

4.4 RDD 数据聚合操作

4.4.1 分组计数

分组计数时 null 不计入统计（非分组时 count 会将 null 计入统计）。

ds.select("tag_code","rule").groupBy("tag_code").count().show()

4.4.2 分组后求最值、平均值、求和

peopleDF.groupBy("address").max("age").show
peopleDF.groupBy("address").avg("age").show
peopleDF.groupBy("address").min("age").show
peopleDF.groupBy("address").sum("age").show

指定字段数据类型（不区分大小写）：

peopleDF.withColumn("count", ds1.col("count").cast("Double")).groupBy("address").max("age").show
peopleDF.withColumn("count", ds1.col("count").cast("double")).groupBy("address").max("age").show

4.4.3 分组后求多个聚合值（groupBy + agg）

peopleDF.groupBy("address").agg(count("age"), max("age"), min("age"), avg("age"), sum("age")).show

4.4.4 分组聚合后取别名

peopleDF.groupBy("address").agg(count("age").as("cnt"), avg("age").as("avg")).show

4.4.5 行转列（pivot 数据透视表）

peopleDF.groupBy("address").pivot("name").avg("age").show
peopleDF.groupBy("address").pivot("name").agg(countDistinct("IdCard").as("uv")).orderBy(col("address")).show
// 提高 pivot 效率：指定对哪些值进行统计汇总
peopleDF.groupBy("address").pivot("name", Seq("lijp", "sunj")).agg(countDistinct("IdCard").as("uv")).orderBy(col("address")).show

4.4.6 不分组求聚合

peopleDF.groupBy().avg("age").show

4.5 SQL 操作

4.5.1 注册表

操作	代码	说明
注册临时表	`df.createOrReplaceTempView("people")`	由当前 SparkSession 持有，Session 消失则临时表销毁。
注册全局表	`df.createGlobalTempView("people")`	由所有 Session 共享，可在新 Session 中继续使用。
执行 SQL	`val sqlDF = spark.sql("SELECT * FROM people")`	—

4.5.2 Join 操作

// 连接（支持 inner、left、right、all）
tag.join(stat1, tag("tag_code")===stat1("tag"), "inner")
// 多次连接
tag.join(stat1, tag("tag_code")===stat1("tag"), "left")
   .join(stat2, tag("tag_code")===stat2("tag"), "left")

选择与重命名：

// 新视图中存在多个重名列，需根据 DF 名称进行选择（否则写入文件会报错）
tag.select(tag("tag"), stat1("tag"))
// 选择多列
tag.select(tag("first_category"), tag("second_category"), tag("tag"), tag("tag_code"))
// 选择并重命名
tag.select(tag("tag").as("tag1"), stat1("tag").as("tag2"))

排序：

// 单列排序
tag.sort("first_category")
// 多列排序
tag.sort("first_category","second_category","tag","tag_code","cover","cover_ratio")

注意：若使用 DataSet 进行 join 操作出现重复列，可使用列名对不同 DataSet 进行索引。结果显示时使用 df.show(false) 防止截断。

4.6 创建 RDD

// 从文件创建
val fileRdd = sc.textFile("/Users/lijp/test.txt")
// 从集合创建
val arrayRdd = sc.parallelize(List((1,1),(2,2)))

4.7 创建 DataFrame

4.7.1 List.toDF

val df = spark.createDataFrame(List(
  ("ming", 20, 15552211521L),
  ("hong", 19, 13287994007L),
  ("zhi", 21, 15552211523L)
)).toDF("name", "age", "phone")

4.7.2 RDD + StructType（推荐）

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
val testRdd = sc.parallelize(List(List(1,1),List(2,2))).map(line => Row(line(0), line(1)))
val schema = StructType(List(StructField("id", IntegerType, true), StructField("code", IntegerType, true)))
val df = spark.createDataFrame(testRdd, schema)

补充：单个元素构成一行

val testRdd = sc.parallelize(List(Row.apply(1), Row.apply(2), Row.apply(3)))
val schema = StructType(List(StructField("id", IntegerType, true)))
val df = spark.createDataFrame(testRdd, schema)

补充：多个元素构成一行（Row.fromSeq）

val testRdd = sc.parallelize(List(Row.fromSeq(List(1,"lijp",99.0)), Row.fromSeq(List(2,"zhangs",85.5)), Row.fromSeq(List(3,"lis",60.0))))
val schema = StructType(List(StructField("id", IntegerType, true), StructField("name", StringType, true), StructField("score", DoubleType, true)))
val df = spark.createDataFrame(testRdd, schema)

4.8 创建复杂类型 DataFrame

4.8.1 Spark 字段类型与 Scala 类型对照

Spark 字段类型	Scala 类型	Scala 表示
`string`	`String`	如：`"1"`
`integer`	`Integer`	如：`1`
`long`	`Long`	如：`1L`
`short`	`Short`	如：`1:Short`
`float`	`Float`	如：`1f`
`double`	`Double`	如：`1d` 或 `1.0`
`array<string>`	`Seq[String]`	如：`Seq("a", "b")`
`map<string,int>`	`Map[String, Integer]`	如：`Map("key" -> 1)`
`struct<string,int>`	嵌套样例类 / 嵌套 Row	见下方示例
`vector`	—	使用 `Vectors.dense`

4.8.2 创建基础类型 DataFrame

val df = List(("1", 1, 1L, 1:Short, 1f, 1d)).toDF("string", "integer", "long", "short","float","double")
// df.printSchema:
//  |-- string: string (nullable = true)
//  |-- integer: integer (nullable = false)
//  |-- long: long (nullable = false)
//  |-- short: short (nullable = false)
//  |-- float: float (nullable = false)
//  |-- double: double (nullable = false)

// 使用 UDF 解析基础类型列
case class MyRow(string: String, integer: Integer, long: Long, short: Short, float: Float, double: Double)
val myudf = udf((x: Row) => MyRow(x.getAs[String]("string"), x.getAs[Integer]("integer"), x.getAs[Long]("long"), x.getAs[Short]("short"), x.getAs[Float]("float"), x.getAs[Double]("double")))
df.withColumn("myudf", myudf(struct(col("string"), col("integer"), col("long"), col("short"), col("float"), col("double")))).show(false)

4.8.3 创建 map 类型 DataFrame

case class MyMap(map: Map[String, Integer])
val df = List(MyMap(Map("1" -> 1))).toDF
// Schema: map: map (key: string, value: integer)

// UDF 解析：map 类型的列使用 Map[T1, T2] 解析
val myudf = udf((x: Map[String, Integer]) => x.keys.toSeq)
df.withColumn("myudf", myudf(col("map"))).show(false)

4.8.4 创建 array 类型 DataFrame

case class MyArray(array: Seq[String])
val df = List(MyArray(Seq("1", "2", "3"))).toDF
// Schema: array: array (element: string)

// UDF 解析：基本类型元素使用 Seq[T]，非基本类型使用 Seq[Row]
val myudf = udf((x: Seq[String]) => x.map(e => e + "$"))
df.withColumn("myudf", myudf(col("array"))).show(false)

4.8.5 创建 struct 类型 DataFrame

从 Spark 角度看 struct 是复合列，从 Scala 角度看 struct 是嵌套 StructType。

方式一：使用 struct 函数创建

// 借助 struct 创建 struct 类型
val df = List(("1",1)).toDF("string", "integer").select(struct(col("string"), col("integer")).as("struct"))
// 可重命名
val df = List(("1",1)).toDF("string", "integer").select(struct(col("string").as("struct_string"), col("integer").as("struct_integer")).as("struct"))

方式二：使用样例类 + Row + StructType 创建

case class Inner(inner_id: Integer, inner_name: String)
case class Outer(outer_id: String, inner: Inner)

val innerStructType = StructType(StructField("inner_id", IntegerType, false) :: StructField("inner_name", StringType, false) :: Nil)
val outerStructType = StructType(StructField("outer_id", StringType) :: StructField("inner", innerStructType) :: Nil)
val rowRdd = sc.parallelize(Seq(Outer("outer", Inner(1, "inner")))).map(outer => Row(outer.outer_id, Row(outer.inner.inner_id, outer.inner.inner_name)))
val df = spark.createDataFrame(rowRdd, outerStructType)

// Schema:
//  |-- id: string
//  |-- mystruct: struct
//  |    |-- struct_ints: integer
//  |    |-- struct_strings: string

// UDF 解析：struct 类型使用 Row 解析
val get_inner_id = udf((x: Row) => x.getAs[Integer]("inner_id"))
df.withColumn("get_inner_id", get_inner_id(col("inner"))).show(false)

4.8.6 复合类型中 Row 和 Struct 的获取

注意：getAs[T] 不支持 Seq[Row] 和 Struct 作为类型参数。

替代方案 1：getSeq[Row]

case class MyStruct(name: String, age: Integer)
case class SeqRow(seq: Seq[MyStruct])
case class NestSeqRow(seqrow: SeqRow)

val df = List(NestSeqRow(SeqRow(Seq(MyStruct("lijp", 35), MyStruct("sunj", 30))))).toDF

// Schema:
//  |-- seqrow: struct
//  |    |-- seq: array
//  |    |    |-- element: struct
//  |    |    |    |-- name: string
//  |    |    |    |-- age: integer

val myudf = udf((x: Row) => x.getSeq[Row](x.fieldIndex("seq")).map(s => s.getAs[String]("name")))
df.withColumn("myudf", myudf(col("seqrow"))).show(false)

替代方案 2：x.getStruct(x.fieldIndex("colname"))

case class InnerStruct(name: String, age: Integer)
case class MiddleStruct(name: String, worker: InnerStruct)
case class OuterStruct(company: String, department: MiddleStruct)

val df = List(OuterStruct("fibodt", MiddleStruct("analyst", InnerStruct("lijp", 35)))).toDF
// Schema:
//  |-- company: string
//  |-- department: struct
//  |    |-- name: string
//  |    |-- worker: struct
//  |    |    |-- name: string
//  |    |    |-- age: integer

val myudf = udf((x: Row) => {
  val innerstruct = x.getStruct(x.fieldIndex("worker"))
  innerstruct.getAs[String]("name")
})
df.withColumn("myudf", myudf(col("department"))).show(false)

4.8.7 根据样例类自动生成 Schema

import org.apache.spark.sql.Encoders
import org.apache.spark.sql.types.StructType

case class Person(name: String, age: Int)
val encoder = Encoders.product[Person]
val structType: StructType = encoder.schema

4.8.8 使用 GenericRowWithSchema 构造默认 Row

用于 UDF 内部提供默认 Row 实例（常见于 if-else、match、map 等匹配操作，防止返回 Any 类型）。

基本用法：

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema

case class Inner(inner_id: Integer, inner_name: String)
case class Outer(outer_id: Integer, outer_name: String, inner: Inner)

val innerStructType = StructType(StructField("inner_id", IntegerType, false) :: StructField("inner_name", StringType, false) :: Nil)
val outerStructType = StructType(StructField("outer_id", IntegerType, false) :: StructField("outer_name", StringType, false) :: StructField("inner", innerStructType, false) :: Nil)

// 由内到外逐步创建 GenericRowWithSchema
val innerRow = new GenericRowWithSchema(Array(222, "inner"), innerStructType)
val outerRow = new GenericRowWithSchema(Array(111, "outer", innerRow), outerStructType)
val defaultRow = outerRow

注意：val outerRow = new GenericRowWithSchema(Row(111, "outer", Row(222, "inner")).toSeq.toArray, outerStructType) 这种方式会导致内层 struct 的 schema 为 null，获取字段时报错。需要改用 getStruct 后重建 GenericRowWithSchema。

应用场景：Full Outer Join 后横向连接

val df1 = List(Outer(1, "one1", Inner(1, "lijp1")), Outer(2, "two1", Inner(2, "zhangsan1")), Outer(3, "three1", Inner(3, "lisi1")), Outer(4, "four1", Inner(4, "wangwu1")), Outer(5, "five1", Inner(5, "zhaoliu1"))).toDF
val df2 = List(Outer(1, "one2", Inner(5, "lijp2")), Outer(2, "two2", Inner(4, "zhangsan2")), Outer(3, "three2", Inner(3, "lisi2")), Outer(4, "four2", Inner(2, "wangwu2")), Outer(5, "five2", Inner(1, "zhaoliu2"))).toDF
val df = df1.join(df2, df1("outer_id") === df2("outer_id"), "fullouter")

// 遍历每一行，比较 inner_id，保留较大的
val merge = udf((df1: Row, df2: Row) => {
  val inner_id1 = df1.getStruct(df1.fieldIndex("inner")).getAs[Integer]("inner_id")
  val inner_id2 = df2.getStruct(df1.fieldIndex("inner")).getAs[Integer]("inner_id")
  val data = if(inner_id1 > inner_id2) df1 else if(inner_id1 < inner_id2) df2 else defaultRow
  Outer(data.getAs[Integer]("outer_id"), data.getAs[String]("outer_name"),
    Inner(data.getStruct(data.fieldIndex("inner")).getAs[Integer]("inner_id"),
      data.getStruct(data.fieldIndex("inner")).getAs[String]("inner_name")))
})

df.withColumn("merge", merge(struct(df1("*")), struct(df2("*")))).select(col("merge.*")).show(false)

4.8.9 使用 typedLit 构造空的 Array 列

场景	代码	说明
错误做法	`df.withColumn("empty_array", lit(Seq()))`	报错：`Unsupported literal type`，Spark 无法确定数据类型。
正确做法 1	`df.withColumn("empty_array", typedLit[Seq[String]](Seq()))`	通过 `typedLit` 显式指定类型。
正确做法 2	`df.withColumn("empty_array", typedLit(Seq.empty[String]))`	通过 `Seq.empty` 显式指定类型。

4.8.10 使用 Vectors 和 VectorUDT 构造 Vector 列

import org.apache.spark.ml.linalg.SQLDataTypes.VectorType
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.types._

val schema = StructType(StructField("vec", VectorType) :: Nil)
val df = spark.createDataFrame(sc.parallelize(List(Row(Vectors.dense(1, 2, 3.0)))), schema)

PySpark 版本：

from pyspark.ml.linalg import VectorUDT, Vectors
from pyspark.sql.types import *
from pyspark.sql.functions import *

schema = StructType([StructField('vector', VectorUDT())])
df = spark.createDataFrame([(Vectors.dense(1.0, 2.0, 3.0),)], schema)
vectorToAny = lambda index: udf(lambda vector: vector[index], StringType())
df.withColumn("test", vectorToAny(0)(col("vector"))).show(truncate=False)

4.9 创建 DataSet

DataFrame 对每个字段的类型要求强一致，DataSet 对每行记录的类型要求强一致。

// 方式一：List.map(beanClass(_))
case class Test(Field1: Int, Field2: Int)
val df = spark.createDataFrame(List(List(1,1),List(2,2),List(3,3)).map(line => Test(line(0),line(1))))

// 方式二：RDD.map(beanClass(_))
case class Test(Field1: Int, Field2: Int)
val testRdd = sc.parallelize(List(List(1,1), List(2,2)))
val df = spark.createDataFrame(testRdd.map(line => Test(line(0), line(1))))

4.10 转化为 DataSet

// 定义样例类
case class Person(name: String, age: Long)
// 由样例类直接创建 DS
val caseClassDS = Seq(Person("Andy", 32)).toDS()
// 由 SparkSQL 产生的 DF 转化为 DS
val peopleDS = spark.read.json(path).as[Person]

4.11 转化为 RDD

4.11.1 由 sc 转化为 DF

import spark.sqlContext.implicits._
import spark.implicits._

case class Line(uid: String, tag: String, time: String, freq: String)
val tagDF = sc.textFile(path).map(line => line.split("\\|")).map(line => Line(line(0), line(1), line(2), line(3))).toDF()

4.11.2 DataSet 操作大全

缓存：

操作	说明
`checkpoint`	为当前 DataSet 设置检查点，数据存储到 checkpoint 目录。
`cache`	将数据集缓存到内存。
`persist`	建立数据集缓存，支持 `MEMORY_ONLY`、`MEMORY_AND_DISK`（默认）等。
`unpersist`	删除缓存。

结构属性：

操作	说明
`columns`	返回列名。
`dtypes`	返回列名和数据类型。
`explain`	返回执行计划。

数据转换：

操作	说明
`rdd`	转换成 RDD。
`toDF`	转换为 DataFrame。

保存文件：

操作	说明
`write`	保存到文件。
`writeStream`	流式数据保存到文件。

创建临时视图：

操作	说明
`createOrReplaceTempView`	创建或替换临时视图。

Action 操作：

操作	说明
`show`	查看数据。
`collect`	以数组形式返回数据集。
`first()`	返回第一行数据。
`head(5)`	返回头部 5 行数据。
`take(5)`	返回头部 5 行数据。

统计数据集：

操作	说明
`count`	返回数据集行数。
`describe`	返回数据集统计信息（count、mean、stddev、min、max）。
`summary`	返回数据集统计信息。
`reduce(func)`	对数据集的每一行执行规约函数。

Transform 操作：

操作	说明
`as` / `alias`	返回具有别名的新数据集。
`map(func)`	对每一行使用函数处理。
`flatMap`	对每一行处理并对结果 explode，每个元素转为一行。
`mapPartitions`	对每个分区使用函数处理。

过滤操作：

操作	示例	说明
`filter`（有类型）	`ds.filter(user => user.name == "user1").show`	使用 Lambda 表达式。
`filter`（无类型）	`ds.filter(col("user") === "user1").show`	使用 Column 表达式。
`where`	同 DataFrame 用法	—

去重操作：

操作	说明
`distinct`	去重。
`dropDuplicates(Seq("col1", "col2"))`	对指定列删除重复数据。

集合操作：

操作	说明
`except(other)`	差集。
`union(other)`	并集。
`intersect(other)`	交集。

排序操作：

操作	说明
`sort("col1", "col2")`	全局排序，支持多列。
`orderBy("col1", "col2")`	全局排序，支持多列。
`sortWithinPartitions("col1", "col2")`	分区内排序。

抽样操作：

操作	说明
`randomSplit(weights, seed)`	按权重分割数据。
`randomSplitAsList(weights, seed)`	同上，返回 List。
`sample(withReplacement, fraction, seed)`	抽样：`withReplacement=true` 且 `fraction>1` 为过采样，`false` 且 `fraction<1` 为负采样。
`sampleBy(colName, fractions, seed)`	按列中各值的抽样概率抽取样本。

调整分区：

操作	说明
`repartition(5)`	重分区为 5 个分区。
`repartition(col("age"))`	根据 age 列的 hash 值分区。
`coalesce`	重分区，默认不 shuffle，适合快速缩减分区。

列操作：

操作	说明
`drop("col1", "col2")`	删除列。
`withColumn`	增加列。
`withColumnRenamed`	列重命名。

连接：join

4.11.3 DataFrame 操作大全

根据索引/键操作：

// 根据索引操作
tagDF.map(tag => "tagcode: " + tag(2)).show()
// 根据键操作
tagDF.map(tag => "tagcode: " + tag.getAs[String]("tagcode")).show()

显示与收集数据：

操作	说明
`show(10, false)`	显示十行，不省略末尾。
`collect()`	收集所有结果，返回 Array。
`collectAsList()`	收集所有结果，返回 List。
`df.describe("user").show()`	显示指定字段的描述统计信息。

判空：

df.isEmpty       // DataFrame 判空
df.rdd.isEmpty   // RDD 判空

获取头部数据：

操作	说明
`first()`	返回第一行。
`head(5)`	返回头部 5 行。
`take(5)`	返回头部 5 行。
`takeAsList()`	返回头部数据为 List。

筛选数据：

操作	示例	说明
`where`	`df.where("user=1 or type ='助手1'").show()`	SQL 字符串。
`filter`	`df.filter("user=1 or type ='助手1'").show()`	SQL 字符串。
`like`	`df.where(data("uid").like("%lijp%")).show()`	模糊匹配，需和 where 一起用。取反：`!data("uid").like("%lijp%")`。对新增列用 `functions.col("uid").like("%lijp%")`。
`in`	`this.where(col("uid").isin(that.select("uid").distinct().collect().map(_.getAs[String]("uid")):_*))`	三步：collect 取回 → map getAs → `:_` 列表解析。也支持 `isin("user1","user2","user3")` 或 `isin(List(...):_)`。
`select`	`df.select("user","type").show()`、`df.select(df("user"),df("user")).show()`	可对字段做简单逻辑处理。
`selectExpr`	`df.selectExpr("user","type as visittype","to_date(visittime)").show()`	传入 UDF、函数、as 别名。
`col` / `apply`	`df.col("user")` 等价于 `df("user")`	—
`limit`	`df.limit(n)`	获取前 n 行，非 Action 操作。
`orderBy` / `sort`	`df.orderBy("visittime").show()`、`df.orderBy(df("visittime").desc).show()`	默认为升序，`.desc` 降序。
`groupBy`	`df.groupBy("user").count().show()`	支持字段名或 Column 对象。
`distinct`	去重。	—
`dropDuplicates`	`df.dropDuplicates("c1","c2")`	存在重复时保留第一行。
`drop`	`df.drop("col1", "col2")`	去除指定字段。
`agg`	`df.groupBy("user").agg(max("id"),sum("user")).show()`、`df.agg("id"->"max","user"->"sum").show()`	—
`withColumn`	见下方示例。	—
`join`	见下方示例。	—
`between`	`df.where(col("visittime").between(1,3))`	范围筛选，包含两端边界。

withColumn 使用示例：

// 添加新列
df.withColumn("sex", df("user") % 2).show()
// 覆盖旧列
df.withColumn("newname", df("oldname")).show()
// 添加常数列
import org.apache.spark.sql.functions.lit
val newdf = df.withColumn("newcol", lit("myval"))

join 使用示例：

// 根据指定列连接（保留唯一字段名）
df.join(df2, Seq("id"))
// 根据指定列连接（保留各自字段名）
df.join(df2, df("id") === df2("id"))

数据统计描述 stat（均为 Action 操作）：

操作	说明
`corr(col1, col2)`	计算两列之间的皮尔逊相关系数。
`cov(col1, col2)`	计算两列之间的协方差。
`freqItems(cols, support)`	查找列的频繁项集，support 为 0~1 的支持度阈值。
`crosstab(col1, col2)`	计算两列之间的交叉表（行名列 1，列名列 2，值为 PV）。
`approxQuantile(col, probabilities, relativeError)`	计算近似分位数。

4.12 聚合函数（UDAF）

4.12.1 不带类型的 UDAF（UserDefinedAggregateFunction）

用于 DataFrame 类型的聚合操作，可配合 Window.partitionBy 分窗函数使用。

示例 1：实现平均值聚合

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._

object avg extends UserDefinedAggregateFunction {
  // 输入数据 Schema
  override def inputSchema: StructType = StructType(StructField("input", LongType) :: Nil)
  // 缓存数据结构（sum 和 count 两个字段）
  override def bufferSchema: StructType = StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)
  // 返回值数据类型
  override def dataType: DataType = DoubleType
  // 幂等性
  override def deterministic: Boolean = true

  // 初始化缓存
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0L
    buffer(1) = 0L
  }

  // 更新缓存（注意：不带类型 UDAF 只能使用索引访问）
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getLong(0) + input.getLong(0)
    buffer(1) = buffer.getLong(1) + 1
  }

  // 合并缓存
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }

  // 计算结果
  override def evaluate(buffer: Row): Any = buffer.getLong(0).toDouble / buffer.getLong(1)
}

使用：

spark.udf.register("my_avg", avg)
val df = List(("lijp", 100), ("lijp", 90), ("lijp", 80), ("sunj", 100), ("sunj", 90), ("sunj", 80)).toDF("name", "score")
df.createOrReplaceTempView("df")
// SparkSQL 方式
spark.sql("select name, my_avg(score) as avg from df group by name").show()
// callUDF 方式
df.groupBy("name").agg(callUDF("my_avg", col("score"))).show(false)

示例 2：countDistinct 聚合函数（支持 Window.partitionBy）

object countDistinct extends UserDefinedAggregateFunction {
  override def inputSchema: StructType = StructType(StructField("input", IntegerType) :: Nil)
  override def bufferSchema: StructType = StructType(StructField("set", ArrayType(IntegerType)) :: StructField("cnt", LongType) :: Nil)
  override def dataType: DataType = LongType
  override def deterministic: Boolean = true

  override def initialize(buffer: MutableAggregationBuffer) = {
    buffer(0) = Array()
    buffer(1) = 0L
  }

  override def update(buffer: MutableAggregationBuffer, input: Row) = {
    if (!buffer.getSeq(0).contains(input.getAs[Integer](0))) {
      buffer(0) = buffer.getSeq(0) :+ input.getAs[Integer](0)
      buffer(1) = buffer.getLong(1) + 1
    } else {
      buffer(0) = buffer.getSeq(0)
      buffer(1) = buffer.getLong(1)
    }
  }

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
    buffer1(0) = buffer1.getSeq(0) ++ buffer2.getSeq(0)
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }

  override def evaluate(buffer: Row) = buffer.getLong(1)
}

示例 3：countMode 聚合函数（多列聚合，众数唯一取众数，否则取最新）

object countMost extends UserDefinedAggregateFunction {
  override def inputSchema: StructType = StructType(StructField("status", StringType) :: StructField("date", IntegerType) :: Nil)
  override def bufferSchema: StructType = StructType(StructField("status_freq", MapType(StringType, LongType)) :: StructField("status_date", MapType(StringType, IntegerType)) :: Nil)
  override def dataType: DataType = StringType
  override def deterministic: Boolean = true

  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = Map[String, Long]()
    buffer(1) = Map[String, Int]()
  }

  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    val status_freq = buffer.getMap[String, Long](0)
    val status = input.getAs[String](0)
    val status_date = buffer.getMap[String, Int](1)
    val date = input.getAs[Int](1)
    if (status_freq.contains(status)) {
      buffer(0) = status_freq.updated(status, status_freq(status) + 1L)
    } else {
      buffer(0) = status_freq.updated(status, 1L)
    }
    if (status_date.contains(status) && status_date(status) < date) {
      buffer(1) = status_date.updated(status, date)
    } else if (!status_date.contains(status)) {
      buffer(1) = status_date.updated(status, date)
    }
  }

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getMap[String, Long](0) ++ buffer2.getMap[String, Long](0)
    buffer1(1) = (buffer1.getMap[String, Int](1).toList ++ buffer2.getMap[String, Int](1).toList)
      .groupBy { case (key, _) => key }
      .map { case (key, values) => (key, values.map(_._2).max) }
  }

  override def evaluate(buffer: Row): Any = {
    val status_freq = buffer.getMap[String, Long](0)
    val status_freq_max = status_freq.maxBy(t => t._2.toInt)._2
    val freq_max_status = status_freq.filter(t => t._2.toInt == status_freq_max.toInt)
    val status_date = buffer.getMap[String, Int](1)
    val check = status_freq.count(t => t._2 == status_freq_max)
    if (check == 1) {
      status_freq.maxBy(t => t._2.toInt)._1
    } else {
      status_date.filter(t => freq_max_status.contains(t._1)).maxBy(t => t._2.toInt)._1
    }
  }
}

示例 4：处理元组作为元素的数组

注意：Spark 的 DataType 不支持元组类型，需用 GenericRowWithSchema 将元组转为 Row。

import org.apache.spark.sql.expressions.{Window, MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema

object statusList extends UserDefinedAggregateFunction {
  val rowSchema = StructType(StructField("status", StringType) :: StructField("date", IntegerType) :: Nil)
  override def inputSchema: StructType = StructType(StructField("status", StringType) :: StructField("date", IntegerType) :: Nil)
  override def bufferSchema: StructType = StructType(StructField("status_date", ArrayType(rowSchema)) :: Nil)
  override def dataType: DataType = ArrayType(StringType)
  override def deterministic: Boolean = true

  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = Seq[Row]()
  }

  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    val status_date = buffer.getSeq[Row](0)
    val status = input.getAs[String](0)
    val date = input.getAs[Int](1)
    // 使用 GenericRowWithSchema 将元组解析为 struct
    val data = new GenericRowWithSchema(Array(status, date), rowSchema)
    buffer(0) = status_date :+ data
  }

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getSeq[Row](0) ++ buffer2.getSeq[Row](0)
  }

  override def evaluate(buffer: Row): Any = {
    buffer.getSeq[Row](0).sortBy(row => row.getAs[Int](1)).map(row => row.getAs[String](0))
  }
}

4.12.2 带有类型的 UDAF（Aggregator）

用于 DataSet 的聚合操作，继承 Aggregator[IN, BUF, OUT]。

import org.apache.spark.sql.{Encoder, Encoders}
import org.apache.spark.sql.expressions.Aggregator

case class Average(var sum: Long, var count: Long)
case class User(name: String, score: Int)

object avg extends Aggregator[User, Average, Double] {

  // 初始化缓存
  override def zero: Average = Average(0L, 0L)

  // 更新缓存
  override def reduce(a: Average, u: User): Average = {
    a.sum += u.score.toLong
    a.count += 1
    a
  }

  // 合并缓存
  override def merge(b1: Average, b2: Average): Average = {
    b1.sum += b2.sum
    b1.count += b2.count
    b1
  }

  // 计算结果
  override def finish(reduction: Average): Double = {
    reduction.sum.toDouble / reduction.count.toDouble
  }

  // 缓存编码器
  override def bufferEncoder: Encoder[Average] = Encoders.product
  // 输出编码器
  override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

使用：

import spark.implicits._
val ds = List(User("lijp", 100), User("lijp", 90), User("lijp", 80), User("sunj", 100), User("sunj", 90), User("sunj", 80)).toDS
// 在 select 中调用
ds.select(avg.toColumn.name("avgInSelect")).show
// 在 groupByKey.agg 中调用
ds.groupByKey(_.name).agg(avg.toColumn.name("avgInAgg")).show
// 在 groupBy.agg 中调用无类型 UDAF
ds.groupBy("name").agg(mean("score").as("untypeUDAF")).show

4.13 常用数据源加载与保存

4.13.1 通用加载方法 `spark.read.load()`

格式	代码示例
Avro	`spark.read.format("avro").load("users.avro")` 或 `spark.read.format("com.databricks.spark.avro").load("users.avro")`
Parquet	`spark.read.format("parquet").load("users.parquet")`
JSON	`spark.read.format("json").load("people.json")`
CSV	`spark.read.format("csv").option("sep", ";").option("inferSchema", "true").option("header", "true").load("people.csv")`
SQL 直接运行	`spark.sql("SELECT * FROM parquet.\`users.parquet`”)`

4.13.2 保存方法 `df.write`

df.select(<sql>).write.format(<输出格式>).mode(<保存类型>).save(<保存路径>)

模式	说明
`error`（默认）	路径已存在时报错退出。
`append`	追加模式，追加到文件末尾。
`overwrite`	覆写模式，覆盖已有数据。
`ignore`	忽略模式，路径已存在时不做任何改动（类似 `CREATE TABLE IF NOT EXISTS`）。

4.13.3 Hive 表相关操作

读取 Hive 表：

spark.table("my_table")

追加写入普通 Hive 表：

// 需要使用 format("Hive") 或 format("hive")
df.select("name").write.format("Hive").mode("append").saveAsTable("default.people")
// mode("append").saveAsTable 等价于 insertInto
df.select("name").write.insertInto("default.people")
// 推荐方式：注册临时表后使用 SQL 插入
df.select("name").createOrReplaceTempView("test")
spark.sql("insert into table default.people select * from test")

注意：追加写入的 df 各列顺序必须与原表一致，否则会导致分区错误。

创建分区 Hive 表：

// 1. 设置 hive.exec.dynamic.partition.mode 为 nonstrict
// 2. 必须显式 partitionBy，否则分区字段会变为普通字段
val df = spark.sql("select * from model_dig.e_age_unique_union_model")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
df.write.partitionBy("p_source", "p_operate", "p_province").saveAsTable("default.df")

四种类型 Hive 表 insert overwrite 差异：

表类型	支持 insert overwrite	说明
Hive 创建的内部表/外部表	✅	覆写分区正常。
`format("hive").saveAsTable` 创建	✅	覆写分区正常。
`create external table` 创建	✅	覆写分区正常。
默认 `saveAsTable` 创建（Parquet 格式）	❌	覆写分区会清空其他分区，需用 truncate table 方式。

Parquet 格式表覆写分区正确方式：

// 1. 删除需要插入的分区
spark.sql("alter table tableName drop if exists partition(p_col1=value1, p_col2=value2)")
// 2. 插入数据至指定分区
df.write.partitionBy("p_col1", "p_col2").mode("append").saveAsTable("tableName")

写入静态分区：

val df1 = spark.sql("select * from model_dig.e_age_unique_union_model")
df1.drop("p_source", "p_operate", "p_province").createOrReplaceTempView("other")
// 追加
spark.sql("insert into table default.df partition(p_source='test', p_operate='test', p_province='test') select * from other")
// 覆盖指定分区
spark.sql("insert overwrite table default.df partition(p_source='test', p_operate='test', p_province='test') select * from other")

写入动态分区：

df1.createOrReplaceTempView("other")
// 追加写入动态分区
spark.sql("insert into table default.df partition(p_source, p_operate, p_province) select * from other")

特别提醒：spark.sql 中 insert overwrite ... partition(...) 会清空所有分区再写入！如需覆写指定分区，需分两步：先 drop partition，再 insert into。

Hive 分区修复：

// 修复分区（spark 方式）
spark.catalog.recoverPartitions("tableName")
// 添加分区（多行 SQL 需使用 paste 模式）
spark.sql("""
alter table tableName
add if not exists
partition (col1 = 'val1', col2 = 'val2')
partition (col1 = 'val3', col2 = 'val4')
""")
// 删除分区
spark.sql("""
alter table tableName
drop if exists
partition (col1 = 'val1', col2 = 'val2'),
partition (col1 = 'val3', col2 = 'val4')
""")

4.13.4 Option 参数

Parquet：

spark.read.option("mergeSchema", "true").parquet("data/test_table")

JSON：

spark.read.option("multiline", "true").json("multi.json")
spark.read.option("charset", "UTF-16BE").json("fileInUTF16.json")

CSV：

参数	示例	说明
`sep`	`.option("sep", "\|")`	列分隔符。
`encoding`	`.option("encoding", "UTF-8")`	编码格式。
`quote`	`.option("quote", "'")`	引号字符。如需剔除引号可用 `.option("quote","\b")`，`\b` 表示字符串边界。
`escape`	`.option("escape", "\\")`	转义字符。
`comment`	`.option("comment", "")`	注释标记。
`header`	`.option("header", "true")`	首行是否为标题。
`ignoreLeadingWhiteSpace`	`.option("ignoreLeadingWhiteSpace", "false")`	忽略前导空格。
`ignoreTrailingWhiteSpace`	`.option("ignoreTrailingWhiteSpace", "false")`	忽略尾部空格。
`nullValue`	`.option("nullValue", "")`	空值表示。
`emptyValue`	`.option("emptyValue", "")`	空字符串表示，使结果显示为 `,...,...`。
`nanValue`	`.option("nanValue", "NaN")`	NaN 值表示。
`positiveInf` / `negativeInf`	`.option("positiveInf", "Inf")`	正/负无穷表示。
`dateFormat`	`.option("dateFormat", "yyyy-MM-dd'T'HH:mm:ss.SSSXXX")`	日期格式。
`maxColumns`	`.option("maxColumns", "20480")`	最大列数。
`maxCharsPerColumn`	`.option("maxCharsPerColumn", "-1")`	每列最大字符数，`-1` 表示无限制。
`mode`	`.option("mode", "PERMISSIVE")`	解析模式。
`multiLine`	`.option("multiLine", "false")`	是否跨行记录。

补充：写入 CSV 时如需剔除字段两侧引号，使用 .option("quote","\b")。如果只是让空值不显示，使用 .option("emptyValue", "")。

Text：

spark.read.option("wholetext","false").text("/path/to/spark/README.md")

4.14 分桶、排序、分区

操作	代码	说明
分桶 + 排序	`peopleDF.write.bucketBy(42, "name").sortBy("age").saveAsTable("fileout")`	`bucketBy` 必须和 `sortBy`、`saveAsTable` 一起使用。
分桶	`usersDF.write.bucketBy(42, "name").saveAsTable("fileout")`	根据值取哈希分桶。
排序	`peopleDF.write.sortBy("age").saveAsTable("fileout")`	生成一个排序文件。
分区	`usersDF.write.partitionBy("favorite_color").format("parquet").save("fileout.parquet")`	根据值分组生成多个文件夹（格式：`字段=值`）。
重分区	`usersDF.repartition(1).write.save("fileout.parquet")`	生成指定数量文件。

4.15 占位符

4.15.1 空数据集

// 空 DataSet
spark.emptyDataSet
// 空 DataFrame
spark.emptyDataFrame

创建带有 Schema 的空 DataFrame：

import org.apache.spark.sql._
import org.apache.spark.sql.types._

val schema = StructType(List(StructField("id", StringType, true), StructField("score", StringType, true)))
val emptyDF = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema)

// 根据非空 DataFrame 创建结构一致的空 DataFrame
val dfNotEmpty = List(("one", 1), ("two", 2), ("three", 3)).toDF("cnt", "int")
val dfEmpty = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], dfNotEmpty.schema)

4.15.2 无穷大表示

类型	表示
Double 负无穷	`Double.NegativeInfinity`
Double 正无穷	`Double.PositiveInfinity`
Float 负无穷	`Float.NegativeInfinity`
Float 正无穷	`Float.PositiveInfinity`

4.15.3 空值 null 的定义

// 错误做法：数据处理时不会报错，但写入文件时报错
// AnalysisException: CSV data source does not support null data type.
df.withColumn("test", lit(null))
// 正确做法：使用 cast 转型
df.withColumn("test", lit(null).cast("string"))

五、特殊操作

5.1 函数操作生成新列

使用 Spark 中的 Scala 环境完成行转列 case when then else：

import org.apache.spark.sql.functions._

5.1.1 UDF 函数

import org.apache.spark.sql.functions.udf
import scala.util.Try
import com.fibodt.encrypt.RuleEncryptUtil

object encrypt {
  val encrypt = ???
}

5.1.2 聚合函数

函数	说明
`avg`	平均值
`collect_list`	聚合指定字段的值到 list
`collect_set`	聚合指定字段的值到 set
`corr`	计算两列的 Pearson 相关系数
`count`	计数
`countDistinct`	去重计数，SQL 中用法：`select count(distinct class)`
`covar_pop`	总体协方差（population covariance）
`covar_samp`	样本协方差（sample covariance）
`first`	分组第一个元素
`last`	分组最后一个元素
`grouping`	grouping_id
`kurtosis`	计算峰态（kurtosis）值
`skewness`	计算偏度（skewness）
`max`	最大值
`min`	最小值
`mean`	平均值
`stddev`	即 `stddev_samp`
`stddev_samp`	样本标准偏差（sample standard deviation）
`stddev_pop`	总体标准偏差（population standard deviation）
`sum`	求和
`sumDistinct`	非重复值求和，SQL 中用法：`select sum(distinct class)`
`var_pop`	总体方差（population variance）
`var_samp`	样本无偏方差（unbiased variance）
`variance`	即 `var_samp`

5.1.3 集合函数

取值：如果一列为 array 类型，可以直接使用 (0) 来进行取值。

函数	说明
`array_contains(column, value)`	检查 array 类型字段是否包含指定元素
`array_position(column, value)`	返回 array 类型字段中第一个指定元素的索引，如果为 0 则表示无此元素
`explode`	展开 array 或 map 为多行
`explode_outer`	同 explode，但当 array 或 map 为空或 null 时，会展开为 null
`posexplode`	同 explode，带位置索引
`posexplode_outer`	同 explode_outer，带位置索引
`from_json`	解析 JSON 字符串为 StructType 或 ArrayType，有多种参数形式
`to_json`	转为 json 字符串，支持 StructType, ArrayType of StructTypes, MapType 或 ArrayType of MapTypes
`get_json_object(column, path)`	获取指定 json 路径的 json 对象字符串。`select get_json_object('{"a":1,"b":2}','$.a')`
`json_tuple(column, fields)`	获取 json 中指定字段值。`select json_tuple('{"a":1,"b":2}','a','b')`
`map_keys`	返回 map 的键组成的 array
`map_values`	返回 map 的值组成的 array
`size`	array 或 map 的长度，需要结合 `withColumn` 使用
`sort_array(e: Column, asc: Boolean)`	将 array 中元素排序（自然排序），默认 asc

注意：当存在嵌套 Array 时，explode 可能会在 Array 中的不同字段之间形成笛卡尔积。

示例：存在字段 hosts，由 host 和 frequency 构成：

name  | hosts
"lijp" | [["host1", 1], ["host2", 2]]

需要将 DataFrame 展开成为 name | host | frequency 的形式：

错误做法（会产生笛卡尔积）：

df.withColumn("host", explode("hosts.host"))
  .withColumn("frequency", explode("hosts.frequency"))

错误结果——由于分两次展开，每次展开会在之前的展开结果上进行处理，因此会生成笛卡尔积：

name   | hosts                          | host    | frequency
"lijp" | [["host1", 1], ["host2", 2]]  | "host1" | 1
"lijp" | [["host1", 1], ["host2", 2]]  | "host1" | 2
"lijp" | [["host1", 1], ["host2", 2]]  | "host2" | 1
"lijp" | [["host1", 1], ["host2", 2]]  | "host2" | 2

正确做法——仅展开一次，获取一一对应的中间结果：

df.withColumn("hosts", explode("hosts"))
  .withColumn("host", col("hosts.host"))
  .withColumn("frequency", col("hosts.frequency"))

正确结果：

name   | hosts             | host    | frequency
"lijp" | ["host1", 1]      | "host1" | 1
"lijp" | ["host2", 2]      | "host2" | 2

5.1.4 时间函数

函数	说明
`add_months(startDate: Column, numMonths: Int)`	指定日期添加 n 月
`date_add(start: Column, days: Int)`	指定日期之后 n 天，e.g. `select date_add('2018-01-01', 3)`
`date_sub(start: Column, days: Int)`	指定日期之前 n 天
`datediff(end: Column, start: Column)`	两日期间隔天数
`current_date()`	当前日期
`current_timestamp()`	当前时间戳，TimestampType 类型
`date_format(dateExpr: Column, format: String)`	日期格式化
`dayofmonth(e: Column)`	日期在一月中的天数，支持 date/timestamp/string
`dayofyear(e: Column)`	日期在一年中的天数，支持 date/timestamp/string
`weekofyear(e: Column)`	日期在一年中的周数，支持 date/timestamp/string
`from_unixtime(ut: Column, f: String)`	时间戳转字符串格式
`from_utc_timestamp(ts: Column, tz: String)`	时间戳转指定时区时间戳
`to_utc_timestamp(ts: Column, tz: String)`	指定时区时间戳转 UTC 时间戳
`hour(e: Column)`	提取小时值
`minute(e: Column)`	提取分钟值
`month(e: Column)`	提取月份值
`quarter(e: Column)`	提取季度
`second(e: Column)`	提取秒
`year(e: Column)`	提取年
`last_day(e: Column)`	指定日期的月末日期
`months_between(date1: Column, date2: Column)`	计算两日期差几个月
`next_day(date: Column, dayOfWeek: String)`	计算指定日期之后的下一个周一、二…，dayOfWeek 区分大小写，只接受 `"Mon"`, `"Tue"`, `"Wed"`, `"Thu"`, `"Fri"`, `"Sat"`, `"Sun"`
`to_date(e: Column, format: String)`	字段类型转为 DateType
`trunc(date: Column, format: String)`	日期截断
`unix_timestamp(s: Column, p: String)`	指定格式的时间字符串转时间戳
`unix_timestamp(s: Column)`	同上，默认格式为 `yyyy-MM-dd HH:mm:ss`
`unix_timestamp()`	当前时间戳（秒），底层实现为 `unix_timestamp(current_timestamp(), yyyy-MM-dd HH:mm:ss)`
`window(timeColumn: Column, windowDuration: String, slideDuration: String, startTime: String)`	时间窗口函数，将指定时间（TimestampType）划分到窗口

to_date 使用注意事项：

to_date 只负责解析字符串，但不能进行格式转换。例如 to_date(lit("20200101"), "yyyyMMdd") 可以运行，但是 to_date(lit("20200101"), "yyyy-MM-dd") 会报错。
date_format 只负责格式转换，但不能进行类型转化。例如 date_format(to_date("20200101","yyyyMMdd"), "yyyy-MM-dd") 可以运行，date_format(lit("20200101"),"yyyyMMdd") 返回 null。
通常情况下 to_date 和 date_format 需要配合使用。

5.1.5 数学函数

函数	说明
`cos`, `sin`, `tan`	计算角度的余弦、正弦、正切
`sinh`, `tanh`, `cosh`	计算双曲正弦、双曲正切、双曲余弦
`acos`, `asin`, `atan`, `atan2`	计算余弦/正弦值对应的角度
`bin`	将 long 类型转为对应二进制数值的字符串。例如 `bin("12")` 返回 `"1100"`
`bround`	舍入，使用 Decimal 的 HALF_EVEN 模式：v > 0.5 向上舍入，v < 0.5 向下舍入，v = 0.5 向最近的偶数舍入
`round(e: Column, scale: Int)`	HALF_UP 模式舍入到 scale 为小数点。v >= 0.5 向上舍入，v < 0.5 向下舍入（即四舍五入）
`ceil`	向上舍入
`floor`	向下舍入
`conv(num: Column, fromBase: Int, toBase: Int)`	转换数值（字符串）的进制
`log(base: Double, a: Column)`	$\log_{base}(a)$
`log(a: Column)`	$\log_e(a)$
`log10(a: Column)`	$\log_{10}(a)$
`log2(a: Column)`	$\log_2(a)$
`log1p(a: Column)`	$\log_e(a+1)$
`pmod(dividend: Column, divisor: Column)`	返回 dividend mod divisor 的正值
`pow(l: Double, r: Column)`	$r^l$（注意 r 是列）
`pow(l: Column, r: Double)`	$r^l$（注意 l 是列）
`pow(l: Column, r: Column)`	$r^l$（注意 r, l 都是列）
`radians(e: Column)`	角度转弧度
`rint(e: Column)`	返回与参数最接近的整数值（double 类型）
`shiftLeft(e: Column, numBits: Int)`	向左位移
`shiftRight(e: Column, numBits: Int)`	向右位移
`shiftRightUnsigned(e: Column, numBits: Int)`	向右位移（无符号位）
`signum(e: Column)`	返回数值正负符号
`sqrt(e: Column)`	平方根
`hex(column: Column)`	转十六进制
`unhex(column: Column)`	逆转十六进制

5.1.6 混杂（Misc）函数

函数	说明
`crc32(e: Column)`	计算 CRC32，返回 bigint
`hash(cols: Column*)`	计算 hash code，返回 int
`md5(e: Column)`	计算 MD5 摘要，返回 32 位十六进制字符串
`sha1(e: Column)`	计算 SHA-1 摘要，返回 40 位十六进制字符串
`sha2(e: Column, numBits: Int)`	计算 SHA-2 摘要，返回 numBits 位十六进制字符串。numBits 支持 224, 256, 384, 512

5.1.7 其他非聚合函数

函数	说明
`abs(e: Column)`	绝对值
`array(cols: Column*)`	多列合并为 array，cols 必须为同类型
`map(cols: Column*)`	将多列组织为 map，输入列必须为 (key, value) 形式，各列的 key/value 分别为同一类型
`bitwiseNOT(e: Column)`	按位取反
`broadcast[T](df: Dataset[T])`	将 df 变量广播，用于实现 broadcast join。如 `left.join(broadcast(right), "joinKey")`
`coalesce(e: Column*)`	返回第一个非空值
`col(colName: String)`	返回 colName 对应的 Column
`column(colName: String)`	`col` 函数的别名
`expr(expr: String)`	解析 expr 表达式，将返回值存于 Column
`greatest(exprs: Column*)`	返回多列中的最大值，跳过 Null
`least(exprs: Column*)`	返回多列中的最小值，跳过 Null
`input_file_name()`	返回当前任务的文件名
`isnan(e: Column)`	检查是否 NaN（非数值）
`isnull(e: Column)`	检查是否为 Null
`lit(literal: Any)`	将字面量（literal）创建一个 Column
`typedLit[T](literal: T)`	将字面量创建一个 Column，支持 Scala 类型如 List, Seq, Map
`monotonically_increasing_id()`	返回单调递增唯一 ID，但不同分区的 ID 不连续。ID 为 64 位整型
`nanvl(col1: Column, col2: Column)`	col1 为 NaN 则返回 col2
`negate(e: Column)`	负数，同 `df.select(-df("amount"))`
`not(e: Column)`	取反，同 `df.filter(!df("isActive"))`
`rand()`	随机数 [0.0, 1.0]
`rand(seed: Long)`	随机数 [0.0, 1.0]，使用 seed 种子
`randn()`	随机数，从正态分布取
`randn(seed: Long)`	同上
`spark_partition_id()`	返回 partition ID
`struct(cols: Column*)`	多列组合成新的 struct column

when 用法：

people.select(
  when(people("gender") === "male", 0)
    .when(people("gender") === "female", 1)
    .otherwise(2)
)

如果没有 otherwise 且 condition 全部没命中，则返回 null。

5.1.8 排序函数

函数	说明
`asc(columnName: String)`	正序
`asc_nulls_first(columnName: String)`	正序，null 排最前
`asc_nulls_last(columnName: String)`	正序，null 排最后
`desc(columnName: String)`	倒序
`desc_nulls_first(columnName: String)`	倒序，null 排最前
`desc_nulls_last(columnName: String)`	倒序，null 排最后

排序示例：

// 方式一：使用函数
df.sort(asc("dept"), desc("age"))

// 方式二：使用 Column 对象
df.orderBy(df("dept").asc, df("age").desc)

5.1.9 字符串函数

函数	说明
`ascii(e: Column)`	计算第一个字符的 ASCII 码
`base64(e: Column)`	Base64 编码
`unbase64(e: Column)`	Base64 解码
`concat(exprs: Column*)`	连接多列字符串
`concat_ws(sep: String, exprs: Column*)`	使用 sep 作为分隔符连接多列字符串
`decode(value: Column, charset: String)`	解码
`encode(value: Column, charset: String)`	编码，charset 支持 `'US-ASCII'`, `'ISO-8859-1'`, `'UTF-8'`, `'UTF-16BE'`, `'UTF-16LE'`, `'UTF-16'`
`format_number(x: Column, d: Int)`	格式化 `'#,###,###.##'` 形式的字符串
`format_string(format: String, arguments: Column*)`	将 arguments 按 format 格式化（printf-style）
`initcap(e: Column)`	单词首字母大写
`lower(e: Column)`	转小写
`upper(e: Column)`	转大写
`instr(str: Column, substring: String)`	substring 在 str 中第一次出现的位置
`length(e: Column)`	字符串长度
`levenshtein(l: Column, r: Column)`	计算两个字符串之间的编辑距离（Levenshtein distance）
`locate(substr: String, str: Column)`	substring 在 str 中第一次出现的位置，位置编号从 1 开始，0 表示未找到
`locate(substr: String, str: Column, pos: Int)`	同上，但从 pos 位置后查找
`lpad(str: Column, len: Int, pad: String)`	字符串左填充（用 pad 字符填充 str 至 len 长度）
`rpad(str: Column, len: Int, pad: String)`	字符串右填充
`ltrim(e: Column)`	剪掉左边的空格、空白字符
`rtrim(e: Column)`	剪掉右边的空格、空白字符
`ltrim(e: Column, trimString: String)`	剪掉左边的指定字符
`rtrim(e: Column, trimString: String)`	剪掉右边的指定字符
`trim(e: Column, trimString: String)`	剪掉左右两边的指定字符
`trim(e: Column)`	剪掉左右两边的空格、空白字符
`regexp_extract(e: Column, exp: String, groupIdx: Int)`	正则提取匹配的组
`regexp_replace(e: Column, pattern: Column, replacement: Column)`	正则替换匹配的部分（参数为列）
`regexp_replace(e: Column, pattern: String, replacement: String)`	正则替换匹配的部分
`repeat(str: Column, n: Int)`	将 str 重复 n 次返回
`reverse(str: Column)`	将 str 反转
`soundex(e: Column)`	计算桑迪克斯代码（soundex code），用于按英语发音来索引姓名，发音相同但拼写不同的单词会映射成同一个码
`split(str: Column, pattern: String)`	用 pattern 分割 str，若想按索引使用返回的 Array 结果，需要使用 `getItem(index)`
`substring(str: Column, pos: Int, len: Int)`	在 str 上截取从 pos 位置开始长度为 len 的子字符串
`substring_index(str: Column, delim: String, count: Int)`	按分隔符提取子字符串。count > 0 返回左边所有内容，count < 0 返回右边所有内容。区分大小写
`translate(src: Column, matchingString: String, replaceString: String)`	把 src 中的 matchingString 全换成 replaceString

5.1.10 特殊类型字段操作

Array<Struct<uid: Int, rating: Float>> 类型字段操作。

5.1.11 UDF 函数（User-Defined Function）

callUDF 调用 UDF：

import org.apache.spark.sql._

val df = Seq(("id1", 1), ("id2", 4), ("id3", 5)).toDF("id", "value")
val spark = df.sparkSession
spark.udf.register("simpleUDF", (v: Int) => v * v)
df.select($"id", callUDF("simpleUDF", $"value"))

udf 定义 UDF（需要打 jar 包时，建议使用 functions.udf 定义 UDF）：

import org.apache.spark.sql.functions._
val simpleUDF: UserDefinedFunction = udf((v: Int) => v * v)
df.select(col("id"), simpleUDF(col("value")))

UDF 处理 Array 类型列（Seq 内部元素为基本类型）

val df = List(
    Seq("lijp", "fibodt")
).toDF("test")

// Seq 中为基本类型，可以直接进行遍历
val names = udf((inputs: Seq[String]) => {
  val result = inputs.map(input => input + "$")
  result
})

df.withColumn("names", names(col("test"))).show(false)

UDF 处理 Array 类型列（Seq 内部元素为自定义类型）

// 定义样例类，用于解析 dataFrame 中的 array<struct<name:string,age:int,phone:array<string>>> 类型字段
case class Person(name: String, age: Int, phone: Seq[String])

// 测试样例
val df = List(
    Seq(Person("lijp", 35, Seq("1234", "2345")), Person("fibodt", 4, Seq("1234", "4321")))
).toDF("test")

// ❌ 错误情形：理论上讲可以使用样例类对 Seq 中的元素进行模式匹配，
// 但实际运行时会有类型转换问题：
// org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to Person
val names = udf((inputs: Seq[Person]) => {
  val result = inputs.map(input => input.name)
  result
})

// ✅ 使用 Spark 原生接口 Row，在取值的同时确定类型
// 注意类型是 Seq[Row]，对于 array<struct<name:string,age:int,phone:array<string>>> 类型的列，
// Row 代表了 struct<name:string,age:int,phone:array<string>>
// 在 Spark 中，Row 可以视为具名元组，用于解析 dataFrame 一行记录中的具名元组结构
val names = udf((inputs: Seq[Row]) => {
  val result = inputs.map(input => input.getAs[String]("name"))
  result
})

df.withColumn("names", names(col("test"))).show(false)

UDF 处理嵌套自定义类型

凡是遇到自定义类型，必须使用 Row 进行解析。

// 定义样例类，用于解析 dataFrame 中的
// array<struct<name:string,age:int,job:struct<name:string,content:array<string>>>> 类型字段
case class Person(name: String, age: Int, job: Job)
case class Job(name: String, content: Seq[String])

// 测试样例
val df = List(
    Seq(
      Person("lijp", 35, Job("work1", Seq("fetch water", "sweep floor"))),
      Person("fibodt", 4, Job("work1", Seq("cook", "boiling water")))
    )
).toDF("test")

// ❌ 错误情形：getAs 中同样不能使用自定义类型
val names = udf((inputs: Seq[Row]) => {
  val jobs = inputs.map(input => input.getAs[Job]("job"))
  val content = jobs.map(job => job.content)
  content
})

// ✅ 使用 Spark 原生接口 Row
val names = udf((inputs: Seq[Row]) => {
  val jobs = inputs.map(input => input.getAs[Row]("job"))
  val content = jobs.map(job => job.getAs[Seq[String]]("content"))
  content.flatten
})

df.withColumn("names", names(col("test"))).show(false)

UDF 同时处理多列

// Person 用于解析 struct<name: string, age: int> 类型列
// Job 用于解析 job: struct<name: string, content: array<string>> 类型列
case class Person(name: String, age: Int)
case class Job(name: String, content: Seq[String])

// 测试样例
val df = List(
    (Person("lijp", 35), Job("work1", Seq("fetch water", "sweep floor"))),
    (Person("fibodt", 4), Job("work1", Seq("cook", "boiling water")))
).toDF("person", "job")

// udf 可以接受多个参数，每个参数分别对应一个列
val check = udf((inputs1: Row, inputs2: Row) => {
  val age = inputs1.getAs[Int]("age")
  val jobs = inputs2.getAs[Seq[String]]("content")
  if (age < 35 && jobs.contains("cook")) "keep" else "dismiss"
})

df.withColumn("check", check(col("person"), col("job"))).show(false)

UDF 处理变长列表

变长列表方便对列进行任意组合。

// Person 用于解析 struct<name: string, age: int> 类型列
// Job 用于解析 job: struct<name: string, content: array<string>> 类型列
case class Person(name: String, age: Int)
case class Job(name: String, content: Seq[String])

// 测试样例：有两个员工角逐同一个岗位，保留那个年龄小于 35 岁的
val df1 = List(
    ("1", Person("lijp", 35), Job("work1", Seq("fetch water", "sweep floor")))
).toDF("jobId", "person", "job")

val df2 = List(
    ("1", Person("sunj", 30), Job("work1", Seq("fetch water", "boiling water")))
).toDF("jobId", "person", "job")

// 当传入参数个数不确定时，可以使用变长参数列表的 udf
// 注意：变长参数列表的 udf 必须先用 def 来定义，然后用 udf 对参数列表进行列表解析
import org.apache.spark.sql.Row

def check_(inputs: Row*) = {
  inputs.filter(input => input.getAs[Row]("person").getAs[Integer]("age") < 35)
    .map(input => input.getAs[Row]("person").getAs[String]("name"))
    .head
}

val check = udf((s: Seq[Row]) => check_(s: _*))

// ❌ 错误情形：务必要注意，udf 中的各个参数一定是列，如果是针对变长参数列表的 udf，
// 务必使用 array 来构造 Seq[Row] 类型的列，而不能将 Seq[Row] 类型误认为是 Seq[Column]
// error: type mismatch: found Seq[Column], required Column
df1.join(df2, df1("jobId") === df2("jobId"))
  .select(check(Seq(struct(df1("person"), df1("*")), struct(df2("person"), df2("*")))).as("winner"))
  .show(false)

// ✅ 使用 array 构造一个 Seq[Row] 类型的列，作为 udf 的参数
// 特别注意：df("*") 会自动进行列表解析并展平，
// 例如 struct(df1("person"), df1("*")) 会等价于 struct(df1("person"), df1("jobId"), df1("person"), df1("job"))
df1.join(df2, df1("jobId") === df2("jobId"))
  .select(check(array(struct(df1("person"), df1("*")), struct(df2("person"), df2("*")))).as("winner"))
  .show(false)

关于 col("*") 的自动列表解析问题：

// 测试数据
val df1 = List(("lijp", "1"), ("ww", "2")).toDF("name", "id")
val df2 = List(("zs", "1"), ("ls", "3")).toDF("name", "id")

// 假设有一个简单的 udf，仅为说明自动列表解析问题
def merge_(rows: Row*) = {
  rows.map(row => row.getAs[String]("name")).mkString(",")
}
val merge = udf((s: Seq[Row]) => merge_(s: _*))

// ✅ 正确方式：df1("*") 等价于 df1("name"), df1("id")，相当于展平的多列
// 即 Seq(df1("name"), df1("id")): _*，必须使用 struct 封装，才能成为一列，才能作为 udf 的一个参数
df1.join(df2, df1("id") === df2("id"), "fullouter")
  .select(merge(array(struct(df1("*")), struct(df2("*")))).as("result"))
  .show(false)
// +-------+
// |result |
// +-------+
// |null,ls|
// |lijp,zs|
// |ww,null|
// +-------+

// ❌ 错误方式：错误的认为 df1("*") 是一列
// 此时虽然外层有 array 封装，但实际上等价于 array(df1("name"), df1("age"), df2("name"), df2("age"))
// 对应类型是 Seq[String]，Scala 会尝试将 String 转换为 Row 类型，发生转换失败：
// java.lang.String cannot be cast to org.apache.spark.sql.Row
df1.join(df2, df1("id") === df2("id"), "fullouter")
  .select(merge(array(df1("*"), df2("*"))).as("result"))
  .show(false)

// array(df1("*"), df2("*")) 实际上的类型是 Seq[String]，重新定义 udf 进行验证如下：
def merge2_(rows: String*) = {
  rows.mkString(",")
}
val merge2 = udf((s: Seq[String]) => merge2_(s: _*))

// 此时正常执行：
df1.join(df2, df1("id") === df2("id"), "fullouter")
  .select(merge2(array(df1("*"), df2("*"))).as("result"))
  .show(false)
// +--------------+
// |result        |
// +--------------+
// |null,null,ls,3|
// |lijp,1,zs,1   |
// |ww,2,null,null|
// +--------------+

柯里化转换为 UDF

通过传入数值类型进行初始化，然后生成 udf 函数用于列的处理。

val df = List("1", "2", "3").toDF("i")

def test(x: Int)(y: String) = x + y.toInt

def test1 = udf(test(1) _)
def test2 = udf(test(2) _)

df.select(test1(col("i")).as("test1"), test2(col("i")).as("test2")).show(false)

特殊场景：延迟初始化，在 withColumn 和 select 中才传入参数进行初始化：

// 方式 1：通过参数列表，向 udf 内部封装的函数传参
val df = List("1", "2", "3").toDF("i")

def test(x: Int)(y: String) = x + y.toInt
def test1(x: Int) = udf(test(x) _)

df.select(test1(1)(col("i"))).show(false)

// 方式 2：通过参数列表，向 udf 内部封装函数传参，结合泛型
// 特别注意：定义函数时可以是泛型函数，但是在转换为 udf 函数时必须显性指定泛型为具体类型
import scala.reflect.runtime.universe.TypeTag

case class RowSeq(s: Seq[String], i: Seq[Int])
val df = List(RowSeq(Seq("1", "2", "3"), Seq(1, 2, 3))).toDF

def test[T: TypeTag](x: Map[T, Double])(y: Seq[T]) = y.maxBy(e => x(e))

// ❌ 错误方式：未显式指定类型，转换为 UDF 函数后，使用时报错 type mismatch
def test0[T: TypeTag](x: Map[T, Double]) = udf(test(x) _)
df.select(test0[String](Map("1" -> 1.0, "2" -> 2.0, "3" -> 3.0))(col("s"))).show(false)

// ✅ 正确方式：显式指定类型后，转换为 UDF 函数
def test1(x: Map[String, Double]) = udf(test(x) _)
df.select(test1(Map("1" -> 1.0, "2" -> 2.0, "3" -> 3.0))(col("s"))).show(false)

def test2(x: Map[Int, Double]) = udf(test(x) _)
df.select(test2(Map(1 -> 1.0, 2 -> 2.0, 3 -> 3.0))(col("i"))).show(false)

5.1.12 窗口函数（排名分析函数）

配合分析函数使用：

import org.apache.spark.sql.expressions.Window
// [排名分析函数, 聚合分析函数].over(Window.partitionBy(colName).orderBy(colName)).alias(newColName)

支持的聚合函数

函数	说明
`cume_dist()`	窗口分区中值的累积分布
`currentRow()`	返回表示窗口分区中当前行的特殊帧边界
`rank()`	排名，排名相等会在名次中留下空位（1, 2, 2, 4）
`dense_rank()`	排名，排名相等不会在名次中留下空位（1, 2, 2, 3）
`row_number()`	行号，排名相等时名次仍然按单调递增序列生成数字（1, 2, 3, 4）
`percent_rank()`	返回窗口分区中行的相对排名（即百分比）
`lag(e: Column, offset: Int, defaultValue: Any)`	返回当前行向前偏移的行，使用时需结合 orderBy
`lead(e: Column, offset: Int, defaultValue: Any)`	返回当前行向后偏移的行，使用时需结合 orderBy
`ntile(n: Int)`	返回有序窗口分区中的分组 id（从 1 到 n）
`first(e: Column, ignoreNulls: Boolean)`	返回窗口分区中的第一个值，`ignoreNull=true` 时忽略 null 值
`last(e: Column, ignoreNulls: Boolean)`	返回窗口分区中的最后一个值
`unboundedPreceding()`	返回表示窗口分区中前一行的特殊帧边界
`unboundedFollowing()`	返回表示窗口分区中最后一行的特殊帧边界
`count(e: Column)`	返回窗口分区中指定记录的数量，null 值不会被计算在内
`countDistinct(e: Column)`	当前不支持该操作，使用 `size(collect_set("colName1").over(Window.partitionBy("colName")))` 替代

开窗函数 `over`

用法	说明
`.over()`	对记录进行分组统计，每行每列都可以返回统计值
`Window.partitionBy()`	对记录进行分组
`Window.partitionBy(colName).orderBy(colName)`	对记录进行分组，在组内进行排序
`Window.partitionBy(colName).rowsBetween(start, end)`	滑动窗口：对分组排序后，仅对行号介于 (当前行号 + start) 和 (当前行号 + end) 之间的行进行聚合操作。start/end 可以是正数（1 = 下一行）、负数（-1 = 上一行）、0（当前行）、`Window.unboundedPreceding`（之前所有行）、`Window.unboundedFollowing`（之后所有行）、`Window.currentRow`
`Window.orderBy(colName).rangeBetween(start, end)`	滑动窗口：对排序列（必须为数值类型）的值进行聚合操作，仅对排序列的值介于 (当前值 + start) 和 (当前值 + end) 之间的行进行聚合操作

特殊分组函数：`rollup()` 与 `cube()`

// rollup：从左向右，从整体到局部，依次分组
df.rollup(a, b, c)
// (1) 首先对 (a, b, c) 进行 group by
// (2) 然后对 (a, b) 进行 group by
// (3) 再对 (a) 进行 group by
// (4) 最后对全表进行汇总操作

// cube：从整体到局部，遍历所有组合进行分组
df.cube(a, b, c)
// (1) 首先对 (a, b, c) 进行 group by
// (2) 然后依次是 (a, b), (a, c), (a), (b, c), (b), (c)
// (3) 最后对全表进行汇总操作

计算累积和

按照累积和进行分段，每段求和值为 50，对 id 按照分组行转列：

// test.txt
score|value|id
10|100|1
20|200|2
30|300|3
40|400|4
50|500|5
10|100|6
20|200|7
30|300|8
40|400|9
50|500|10
10|100|11
20|200|12
30|300|13
40|400|14
50|500|15

import org.apache.spark.sql.expressions.Window

val test = spark.read.option("header", "true").option("sep", "|").csv("test.txt")
val group = test
  .withColumn("id", col("id").cast("int"))
  .withColumn("group", ceil(
    sum("score").over(Window.orderBy("id").rowsBetween(Window.unboundedPreceding, Window.currentRow)) / 50
  ))

group.show(false)

val groupLine = group
  .groupBy("group")
  .agg(concat_ws(" or ", collect_set("id")).as("groupLine"))

groupLine.show(false)

`when` 中使用 `over partitionBy`

示例 1：统计用户登录前未登录次数、未登录前登录次数：

val df = Seq(
  ("user1", "login", "20230730"),
  ("user1", "login", "20230731"),
  ("user1", "login", "20230801"),
  ("user1", "nologin", "20230802"),
  ("user1", "nologin", "20230803"),
  ("user1", "nologin", "20230804"),
  ("user1", "nologin", "20230805"),
  ("user1", "login", "20230806"),
  ("user1", "nologin", "20230807"),
  ("user1", "login", "20230808"),
  ("user1", "nologin", "20230809")
).toDF("user", "status", "time")

// 两个步骤：
// 第 1 步：对状态切换的情况进行标注，若状态发生过切换，则为 1
// 第 2 步：对状态未发生切换的情况进行标注（例如连续登录），若状态未发生过切换，则单调递增数列填充
// 注意：无论 when 的条件是否满足，over partitionBy 始终会对全量数据进行计算
// 注意：row_number 函数默认从 1 开始，当分区排序数列中第 1 项有值时，自动从 2 开始递增
df.withColumn("status_seq",
  when(col("status") !== lag(col("status"), 1).over(Window.partitionBy("user").orderBy("time")), lit(1))
)
.withColumn("status_seq",
  when(col("status_seq").isNull,
    row_number().over(Window.partitionBy("user", "status").orderBy("time"))
  ).otherwise(col("status_seq"))
)
.orderBy("user", "time")
.show()

示例 2：统计用户登录天数：

val df = Seq(
  ("user1", "login", "20230730"),
  ("user1", "login", "20230731"),
  ("user1", "login", "20230801"),
  ("user1", "nologin", "20230802"),
  ("user1", "nologin", "20230803"),
  ("user1", "nologin", "20230804"),
  ("user1", "nologin", "20230805"),
  ("user1", "login", "20230806"),
  ("user1", "nologin", "20230807"),
  ("user1", "login", "20230808"),
  ("user1", "nologin", "20230809")
).toDF("user", "status", "time")

// 两个步骤：
// 第 1 步：对状态进行标注，若状态为 login 则为 1，若状态为 nologin 则为 0
// 第 2 步：对标注结果进行求和
df.withColumn("stats_seq",
  sum(
    when(col("status") === "login", lit(1)).otherwise(lit(0))
  ).over(Window.partitionBy("user"))
).show()

5.1.13 使用 DataFrame API 实现矩阵乘法

矩阵乘法计算过程（$A_{ij} \times B_{jk}$）：

将 A 做转置，成为 $A_{ji}$

$A_j$ 和 $B_j$ 对应行进行笛卡尔积组合

在每个组合内，生成 $(i, k)$ 坐标，并计算对应坐标上的临时结果 $v \times w$

丢弃索引 j，根据 $(i, k)$ 坐标进行 groupBy，对汇总后临时结果 $v \times w$ 求和

结果解读：$(i, k)$ 表示最终矩阵坐标，$\sum(v \times w)$ 即该坐标上的对应值

import org.apache.spark.sql.functions._

val m = Seq(
  (0, 0, 1.0), (0, 1, 2.0),
  (1, 0, 3.0), (1, 1, 4.0)
).toDF("i", "j", "v") // 定义矩阵 M

val n = Seq(
  (0, 0, 5.0), (0, 1, 6.0),
  (1, 0, 7.0), (1, 1, 8.0)
).toDF("j", "k", "w") // 定义矩阵 N

val res = m.select(col("j"), struct(col("i"), col("v")).as("m"))
  .join(n.select(col("j"), struct(col("k"), col("w")).as("n")), Seq("j"))
  .withColumn("v", expr("m.v * n.w"))
  .groupBy("m.i", "n.k").agg(sum("v").as("res"))

res.show() // 输出结果

5.2 对新列进行处理

5.2.1 导入隐式转换

必须导入隐式转换，否则使用 $ 选择列时会报错：

import spark.implicits._

5.2.2 使用 `$` 对列值进行筛选

import org.apache.spark.sql.functions._

// 注意：如果用 $("new_tag") 会报错"没有这一列"，需要用 $"new_tag"
s.withColumn("new_tag", lit("test")).where($"new_tag" === "test").show()

5.3 对重名列进行处理

// join 时使用 df("colName") 方式指定列，避免歧义
df1.join(df2, df1("id") === df2("id"), "inner")

5.4 对列进行重命名

df.withColumnRenamed("id", "id_other")

六、复杂数据类型

6.1 Struct

6.1.1 创建 Struct 类型的列

元组类型数据一律优先考虑使用 Struct 类型构造并解析。

注意：

原数据中的 struct 类型字段最好使用 Tuple，可以自动解析内部元素类型，如果使用 Seq 或 List，内部元素类型为 Any。

struct 字段对应的数据应当使用 Row 封装，并且需要对元素进行序列解包，不然会报错。

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

val data = List(
  ("lijp", 35, ("lijp", 35)),
  ("sunj", 30, ("sunj", 30))
).map(line => Row(line._1, line._2, Row(line._3._1, line._3._2)))

val schema = StructType(List(
  StructField("name", StringType),
  StructField("age", IntegerType),
  StructType(List(
    StructField("name", StringType),
    StructField("age", IntegerType)
  ))
))

val df = spark.createDataFrame(sc.parallelize(data), schema)
df.show()

6.1.2 增加 Struct 类型的列

val structDF = df.withColumn("structCol", struct(col("name"), col("age")))

6.1.3 Struct 列的取值

// 使用 getField 方法
structDF.select(col("structCol").getField("name")).show

// 使用点号取指定字段
structDF.select("structCol.name").show

// 使用星号取全部字段
structDF.select("structCol.*").show
// 等价于
structDF.select("structCol.name", "structCol.age").show

6.2 Array

6.2.1 创建 Array 类型的列

注意：原数据中的 array 类型字段使用 List，如果是 Tuple，会被识别为多个 string 类型字段。

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

// 方式一：手动构造 Schema
val data = List(List("lijp", "sunj"), List("zhangsan", "lisi")).map(line => Row(line))
val schema = StructType(List(StructField("names", ArrayType(StringType, true))))
val df = spark.createDataFrame(sc.parallelize(data), schema)

// 方式二：简单方法
val df = List(List("lijp", "sunj"), List("zhangsan", "lisi")).toDF("names")

6.2.2 新增 Array 类型的列

val arrayDF = df.withColumn("arrayCol", array(col("names")(0), col("names")(1)))

6.2.3 Array 列的取值

// 使用 getItem 方法
df.select(col("names").getItem(0).as("name")).show

// 使用方括号索引
df.select(expr("names[0]")).show

// 使用圆括号索引
df.select(col("names")(0).as("name")).show

// 使用 element_at，按索引取值，从 1 开始，超出索引的部分返回 null，支持倒序索引
// 当指定索引为 0 时报错
df.select(element_at(col("names"), 1))

6.2.4 行转列

// 将 array 字段展开成为列
df.withColumn("newname", explode(col("names")))

// 将 array 字段展开为列，当 array 字段为 null 或空时，展开为 null
df.withColumn("newname", explode_outer(col("names")))

6.2.5 常用方法

函数	说明
`array_contains(col("names"), "lijp")`	判断每行记录是否包含指定元素
`array_distinct(col("names"))`	对 array 类型的字段进行去重
`array_except(col("names"), col("other"))`	对两个 array 字段取差集
`array_intersect(col("names"), col("other"))`	对两个 array 字段取交集
`array_union(col("names"), col("other"))`	对两个 array 字段取并集
`array_join(col("names"), "-")`	将 array 字段中的元素以指定字符进行拼接
`array_max(col("names"))`	array 字段排序取最大值
`array_min(col("names"))`	array 字段排序取最小值
`array_position(col("names"), "lijp")`	在 array 字段中寻找指定值的索引，从 1 开始计数，未找到返回 0
`array_remove(col("names"), "lijp")`	在 array 字段中剔除指定元素
`array_repeat(col("names"), 2)`	将原 array 字段作为元素，重复指定次数，形成新的嵌套 array 列
`array_sort(col("names"))`	对 array 字段排序
`arrays_overlap(col("names"), col("other"))`	对两个 array 字段取交集，若交集不为空则返回 true
`arrays_zip(col("names"), col("other"))`	对多个 array 字段做拉链操作，对应位置元素组合为 array。若多个 array 长度不一致，返回新的 array 长度与最长 array 相同，缺少的值位置为空

6.3 Map

6.3.1 创建 Map 类型的列

注意：原数据中的 map 类型字段使用 Map。

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

// 方式一：手动构造 Schema
val data = List(Map("lijp" -> 35), Map("zhangsan" -> 30)).map(line => Row(line))
val schema = StructType(List(StructField("names", MapType(StringType, IntegerType, true))))
val df = spark.createDataFrame(sc.parallelize(data), schema)

// 方式二：简单方法
val df = List(Map("lijp" -> 35), Map("zhangsan" -> 30)).toDF("names")

6.3.2 新增 Map 类型的列

val df = List(("lijp", 35), ("zhangsan", 30)).toDF("name", "age")
df.withColumn("mapCol", map(col("name"), col("age")))

6.3.3 Map 类型的列取值

// 使用方括号，传入列名，进行索引
df.select(expr("mapCol[name]")).show

// 使用方括号，传入列值，进行索引（如果不存在指定的键，则返回 null）
df.select(expr("mapCol['lijp']")).show

// 使用圆括号索引
df.select(col("mapCol")("lijp")).show

// 使用点号索引（不支持星号索引）
df.select(col("mapCol.lijp")).show

// 使用 getField 索引
df.select(col("mapCol").getField("lijp")).show

// 使用 getItem 索引
df.select(col("mapCol").getItem("lijp")).show

6.3.4 行转列

// 将 map 字段展开成列，键为 key 列、值为 value 列
df.select(col("*"), explode(col("mapCol"))).show

// 使用 element_at，按键取值
df.select(element_at(col("names"), "lijp")).show
df.select(element_at(col("names"), col("name"))).show

6.3.5 常用方法

map_from_arrays、map_from_entries、map_concat 从 Spark 2.4 以后引入。

函数	说明
`map_from_arrays(array<K>, array<V>): map<K,V>`	使用两个 array 类型的列，生成一个 Map 类型的列
`map_from_entries(array<struct<K,V>>): map<K,V>`	使用一个 array 类型的列（内部元素为二元 struct 类型），生成一个 Map 类型的列
`map_concat(map<K,V>, ...): map<K,V>`	合并多个 Map 类型列的并集，键不会相互覆盖。注意必须是可变参数列表，如果是 `array<Map<K,V>>` 类型则会报错，该函数不能在 agg 中作为聚合函数使用
`map_keys(col("names"))`	返回键构成的数组
`map_values(col("names"))`	返回值构成的数组

注意：Spark 2.3 及更早版本，可通过以下方式实现 map_from_arrays——使用 map 字段聚合后仍为 map 字段：
df.withColumn("mapCol", map(col("name"), col("age")))
  .groupBy("id").agg(collect_list(col("mapCol")).as("maps"))
  .printSchema

6.4 JSON

JSON 在 Spark 中较为特殊，没有 JSON 类型，是以字符串的形式存储。换言之，利用 JSON 字符串的特性，可以使用 from_json 解析任意复杂结构的字符串类型（如 struct、array、map 等）。

假设有 df 中存在一个 json 字符串的列 jsonstr：

val df = List("""{"role": "robot", "saying": "hello!"}""").toDF("jsonstr")

6.4.1 `get_json_object` 解析 JSON 字符串

path 使用 $ 开头，点号分隔：

df.withColumn("role", get_json_object(col("jsonstr"), "$.role"))
  .withColumn("saying", get_json_object(col("jsonstr"), "$.saying"))

6.4.2 `json_tuple` 解析 JSON 字符串中的单个字段

df.withColumn("robot", json_tuple(col("jsonstr"), "role"))
  .withColumn("saying", json_tuple(col("jsonstr"), "saying"))

6.4.3 `from_json` 根据 Schema 解析 JSON 字符串

import org.apache.spark.sql.types._

val schema = StructType(List(
  StructField("role", StringType),
  StructField("saying", StringType)
))

df.withColumn("jsonstruct", from_json(col("jsonstr"), schema))

6.4.4 `to_json` 将多列转为 JSON 字符串

// 将多个列转成 json 字符串的列
df.select(to_json(struct(col("role"), col("saying"))))

6.4.5 `from_json` 解析复杂结构型数据

注意：JSON 字符串中不支持 -> 符号，[] 中不支持 :。

（1）Struct 类型

val df = List("""{"role": "robot", "saying": "hello!"}""").toDF("jsonstr")
val schema = StructType(List(StructField("role", StringType), StructField("saying", StringType)))
df.withColumn("jsonstr", from_json(col("jsonstr"), schema)).show(false)

（2）Array 类型

val list_df = List("""["1", "2", "3"]""").toDF("list")
val schema = ArrayType(StringType)
list_df.withColumn("list", from_json(col("list"), schema)).show(false)

（3）Map 类型

类型必须一致，否则解析结果为 null。

// ✅ 正确
val map_df = List("""{"one" : "1", "two" : "2", "three" : "3"}""").toDF("map")
val schema = MapType(StringType, StringType)
map_df.withColumn("map", from_json(col("map"), schema)).show(false)

val map_df = List("""{"one" : 1, "two" : 2, "three" : 3}""").toDF("map")
val schema = MapType(StringType, IntegerType)
map_df.withColumn("map", from_json(col("map"), schema)).show(false)

// ❌ 无法解析，需要将 '->' 替换为 ':'，转为 struct 类型处理
val map_df = List("""{"one" -> 1, "two" -> 2, "three" -> 3}""").toDF("map")

// ❌ 无法解析，需要将 '[]' 替换为 '{}'，转为 struct 类型处理
val map_df = List("""["one" : "1", "two" : "2", "three" : "3"]""").toDF("map")

（4）复合类型

Array[Map] 复合：

val array_map_df = List("""[{"one" : 1, "two" : 2, "three" : 3}, {"one" : 11, "two" : 22, "three" : 33}]""").toDF("array_map")
val schema = ArrayType(MapType(StringType, IntegerType))
array_map_df.withColumn("array_map", from_json(col("array_map"), schema)).show(false)

Array[Struct] 复合：

val array_struct_df = List("""[{"name" : "lijp", "age" : 35}, {"name" : "sunj", "age" : 30}]""").toDF("array_struct")
val schema = ArrayType(StructType(List(StructField("name", StringType), StructField("age", IntegerType))))
array_struct_df.withColumn("array_struct", from_json(col("array_struct"), schema)).show(false)

Array[Array] 复合：

val array_array_df = List("""[["one", "two", "three"], ["1" , "2", "3"]]""").toDF("array_array")
val schema = ArrayType(ArrayType(StringType))
array_array_df.withColumn("array_array", from_json(col("array_array"), schema)).show(false)

Map[String, Array] 复合：

val map_array_df = List("""{"count": ["one", "two", "three"], "num": ["1" , "2", "3"]}""").toDF("map_array")
val schema = MapType(StringType, ArrayType(StringType))
map_array_df.withColumn("map_array", from_json(col("map_array"), schema)).show(false)

Map[String, Struct] 复合：

val map_struct_df = List("""{"count": {"scale":3, "number":"three"}}""").toDF("map_struct")
val schema = MapType(StringType, StructType(List(StructField("scale", IntegerType), StructField("number", StringType))))
map_struct_df.withColumn("map_struct", from_json(col("map_struct"), schema)).show(false)

Map[String, Map[String, Int]] 复合：

val map_map_df = List("""{"lijp": {"age":3}, "sunj": {"age": 3}}""").toDF("map_map")
val schema = MapType(StringType, MapType(StringType, IntegerType))
map_map_df.withColumn("map_map", from_json(col("map_map"), schema)).show(false)

Struct[Array] 复合：

val struct_array_df = List("""{"names": ["lijp", "sunj"], "ages": [35, 30]}""").toDF("struct_array")
val schema = StructType(List(
  StructField("names", ArrayType(StringType)),
  StructField("ages", ArrayType(IntegerType))
))
struct_array_df.withColumn("struct_array", from_json(col("struct_array"), schema)).show(false)

Struct[Map] 复合：

val struct_map_df = List("""{"names": {"lijp": "sunj", "sunj": "lijp"}}""").toDF("struct_map")
val schema = StructType(List(
  StructField("names", MapType(StringType, StringType))
))
struct_map_df.withColumn("struct_map", from_json(col("struct_map"), schema)).show(false)

Struct[Struct] 复合：

val struct_struct_df = List("""{"names": {"name1": "lijp", "name2": "sunj"}}""").toDF("struct_struct")
val schema = StructType(List(
  StructField("names", StructType(List(
    StructField("name1", StringType),
    StructField("name2", StringType)
  )))
))
struct_struct_df.withColumn("struct_struct", from_json(col("struct_struct"), schema)).show(false)

七、访问操作元信息

7.1 创建外部表

仅支持 parquet 文件，且仅能创建外部表：

spark.catalog.createTable("test", "/user/lijp/url_parquet")
spark.catalog.createExternalTable("test", "/user/lijp/url_parquet")

7.2 判断是否存在

方法	说明
`spark.catalog.databaseExists`	判断数据库是否存在
`spark.catalog.tableExists`	判断表是否存在
`spark.catalog.functionExists`	判断 UDF 函数是否存在

7.3 返回信息 DataFrame

方法	说明
`spark.catalog.listDatabases`	返回数据库信息
`spark.catalog.listTables`	返回表信息
`spark.catalog.listColumns`	返回列信息
`spark.catalog.listFunctions`	返回函数信息

7.4 建立或删除缓存

方法	说明
`spark.catalog.cacheTable`	缓存表
`spark.catalog.clearCache`	清空缓存
`spark.catalog.isCached`	查看是否被缓存

7.5 删除临时视图、表

注意：spark.catalog 无法删除表，如需删表仍然使用 spark.sql("drop table if exists test")。

方法	说明
`spark.catalog.dropGlobalTempView`	删除全局临时视图
`spark.catalog.dropTempView`	删除临时视图
`spark.sqlContext.dropTempTable`	删除临时表

八、报错处理

缺失已有字段

错误信息：

Exception in thread "main" org.apache.spark.sql.AnalysisException:
resolved attribute(s) dt#1529 missing from ...

原因：DataFrame 进行复杂内连接，字段循环依赖引起的。

val D = A.join(B, Seq("col1"))
  .join(C, A("col1") === C("col1"), "left")
  .where(C("col2").isNotNull)
// 此时虽然字段名称没有改变，字段 col1 仍然是 A 中的字段 col1，
// 但是却已经被其他字段的计算改变

A.join(D, Seq("col1"))
// 此时可能会找不到 A 中的对应字段 col1，
// 应当对 D 中处理完的 col1 指定别名，形成一个新的字段

解决：对 D 中生成字段使用 as 指定别名即可。

九、Spark 配置详解

参考官方文档：Spark Configuration

问题 1：如何处理小文件过多的问题

方案一：使用 coalesce 合并小文件

coalesce 操作可以将多个小文件合并为一个大文件，减少小文件的数量。

注意：coalesce 操作只能用于减少文件的数量，不能用于增加文件的数量。如果想增加文件的数量，可以使用 repartition 操作。

rdd.coalesce(10).saveAsTextFile(outputPath)

方案二：设置 spark.sql.shuffle.partitions

在 Spark 作业的配置中设置该参数来控制 Spark 写文件时生成的文件数量，默认值是 200。

val conf = new SparkConf().set("spark.sql.shuffle.partitions", "100")
val spark = SparkSession.builder.config(conf).getOrCreate()

方案三：使用自定义 Partitioner

通过实现 org.apache.spark.Partitioner 接口，将其传递给 saveAsHadoopFile 或 saveAsNewAPIHadoopFile 方法来控制文件数量。

问题 2：覆写原文件夹的数据

使用 saveAsTextFile 方法覆盖原始文件夹中的数据：

rdd.saveAsTextFile("/path/to/output")

注意：如果原始文件夹中有许多文件，Spark 会在文件夹中创建多个文件来保存数据。因此，执行 saveAsTextFile 操作后，可能会看到原始文件夹中的文件数量发生了变化。

问题 3：如何执行、加载 Scala 脚本

注意：-i < filename1 > filename2 实际上是以 filename1 文件作为标准输入，以 filename2 文件作为标准输出。注意：每一行必须是完整语句，否则会因为分开执行而报错，需删除所有空行，所有换行符 \n 需要用分号代替（需要检查 match）。

命令	说明
`spark-shell -i test.scala`	执行后进入交互命令行 REPL
`spark-shell -i < test.scala`	执行后退出交互命令行 REPL
`spark-shell -i < test.scala > test.log 2>&1`	静默执行，日志重定向写入到文件 test.log
`spark-shell` 启动后 `:load test.scala`	启动交互命令行 REPL 并加载执行脚本

注意：后台执行 spark-shell 不支持，必须使用 spark-submit。

问题 4：Avro 文件损坏