Python装饰器完全指南：从原理到实战

Linux Shell 是与操作系统交互的强大工具。无论是系统管理、文件处理还是网络配置，Shell 命令都能让任务变得简单而高效。本文将整理一些常用的 Linux Shell 命令，帮助大家更好地掌握和使用 Linux。

## 基本命令

### 1. `ls` - 列出目录内容

列出目录中的文件和子目录。

```bash
ls
ls -l    # 详细信息
ls -a    # 包括隐藏文件
```

### 2. `cd` - 切换目录

切换到指定目录。

```bash
cd /path/to/directory
cd ~    # 切换到主目录
cd -    # 切换到上一个目录
```



### 3. `pwd` - 显示当前目录

显示当前工作目录的完整路径。

```bash
pwd
```

### 4. `mkdir` - 创建目录

创建新目录。

```bash
mkdir new_directory
mkdir -p /path/to/directory    # 创建多级目录
```

### 5. `rmdir` - 删除目录

删除空目录。

```bash
rmdir directory_name
```

### 6. `rm` - 删除文件或目录

删除文件或目录（注意：操作不可逆）。

```bash
rm file_name
rm -r directory_name    # 删除目录及其内容
rm -f file_name         # 强制删除文件
```

### 7. `cp` - 复制文件或目录

复制文件或目录。

```bash
cp source_file destination_file
cp -r source_directory destination_directory    # 复制目录及其内容
```

### 8. `mv` - 移动或重命名文件

移动或重命名文件或目录。

```bash
mv old_name new_name
mv /path/to/source /path/to/destination
```

## 文件查看与编辑

### 1. `cat` - 连接并显示文件内容

显示文件内容。

```bash
cat file_name
```

### 2. `less` - 分页查看文件内容

分页显示文件内容，便于长文件阅读。

```bash
less file_name
```

### 3. `head` - 查看文件开头部分

显示文件的前几行。

```bash
head file_name
head -n 20 file_name    # 显示前20行
```

### 4. `tail` - 查看文件末尾部分

显示文件的后几行。

```bash
tail file_name
tail -n 20 file_name    # 显示后20行
```

### 5. `nano` 和 `vim` - 文件编辑器

编辑文件。

```bash
nano file_name
vim file_name
```

## 文件搜索与查找

### 1. `find` - 搜索文件或目录

根据条件搜索文件或目录。

```bash
find /path -name "file_name"
find /path -type d -name "directory_name"    # 搜索目录
```

### 2. `grep` - 搜索文本内容

在文件中搜索匹配的文本。

```bash
grep "search_term" file_name
grep -r "search_term" /path    # 递归搜索
```

## 系统信息

### 1. `uname` - 显示系统信息

显示操作系统的相关信息。

```bash
uname
uname -a    # 显示详细信息
```

### 2. `top` - 动态显示系统进程

实时显示系统进程信息。

```bash
sh复制代码
top
```

### 3. `ps` - 显示当前进程

显示当前用户的进程列表。

```bash
ps
ps aux    # 显示所有进程
```

### 4. `df` - 显示磁盘使用情况

显示文件系统的磁盘空间使用情况。

```bash
df -h    # 以人类可读的格式显示
```

### 5. `du` - 显示目录空间使用情况

显示目录及其子目录占用的磁盘空间。

```bash
du -sh /path/to/directory
```

## 网络命令

### 1. `ping` - 检测网络连接

检测与目标主机的网络连接情况。

```bash
ping www.example.com
```

### 2. `ifconfig` - 配置网络接口

显示或配置网络接口（现代系统多用 `ip` 命令）。

```bash
ifconfig
```

### 3. `curl` - 发送网络请求

用来传输数据。

```bash
curl http://www.example.com
```

## 权限与所有权

### 1. `chmod` - 修改文件权限

更改文件或目录的权限。

```bash
chmod 755 file_name
chmod -R 755 directory_name    # 递归更改目录权限
```

### 2. `chown` - 更改文件所有权

更改文件或目录的所有者和所属组。

```bash
chown user:group file_name
chown -R user:group directory_name    # 递归更改目录所有权
```

## 结语

掌握 Linux Shell 的常用命令可以显著提高你的工作效率。以上列出的命令只是其中的一部分，Linux 提供了丰富的命令和工具，持续学习和实践将帮助你更好地利用这些工具来解决实际问题。

希望这篇文章能帮助你更好地理解和使用 Linux Shell。如果你有任何疑问或建议，欢迎在评论区留言。

Linux Shell(常用命令整理)

## 通用的加载和保存方式

SparkSQL 提供了通用的保存数据和数据加载的方式。这里的通用指的是使用相同的 API，根据不同的参数读取和保存不同格式的数据，SparkSQL 默认读取和保存的文件格式 为 parquet

1. 加载数据

spark.read.load 是加载数据的通用方法

```scala
scala> spark.read.
csv	format jdbc json load option options orc parquet schema table text textFile
```

如果读取不同格式的数据，可以对不同的数据格式进行设定

```scala
scala> spark.read.format("…")[.option("…")].load("…")
```

- format("…")：指定加载的数据类型，包括"csv"、"jdbc"、"json"、"orc"、"parquet"和 "textFile"。
- load("…")：在"csv"、"jdbc"、"json"、"orc"、"parquet"和"textFile"格式下需要传入加载 数据的路径。
- option("…")：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable 我们前面都是使用 read API 先把文件加载到 DataFrame 然后再查询，其实，我们也可以直 接在文件上进行查询: 文件格式.`文件路径`



```scala
scala>spark.sql("select * from json.`/opt/module/data/user.json`").show
```

1. 保存数据

df.write.save 是保存数据的通用方法

```scala
scala>df.write.
csv jdbc json orc parquet textFile… …
```

如果保存不同格式的数据，可以对不同的数据格式进行设定

```scala
scala>df.write.format("…")[.option("…")].save("…")
```

- format("…")：指定保存的数据类型，包括"csv"、"jdbc"、"json"、"orc"、"parquet"和 "textFile"。
- save ("…")：在"csv"、"orc"、"parquet"和"textFile"格式下需要传入保存数据的路径。
- option("…")：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable 保存操作可以使用 SaveMode, 用来指明如何处理数据，使用 mode()方法来设置。 有一点很重要: 这些 SaveMode 都是没有加锁的, 也不是原子操作。

SaveMode 是一个枚举类，其中的常量包括：

| Scala/Java | Any Language | Meaning |
| --- | --- | --- |
| SaveMode.ErrorIfExists(default) | "error"(default) | 如果文件已经存在则抛出异常 |
| SaveMode.Append | "append" | 如果文件已经存在则追加 |
| SaveMode.Overwrite | "overwrite" | 如果文件已经存在则覆盖 |
| SaveMode.Ignore | "ignore" | 如果文件已经存在则忽略 |

```scala
df.write.mode("append").json("/opt/module/data/output")
```

## Parquet

Spark SQL 的默认数据源为 Parquet 格式。Parquet 是一种能够有效存储嵌套数据的列式 存储格式。

数据源为 Parquet 文件时，Spark SQL 可以方便的执行所有的操作，不需要使用 format。 修改配置项spark.sql.sources.default，可修改默认数据源格式。

1. 加载数据

```scala
scala> val df = spark.read.load("examples/src/main/resources/users.parquet")
scala> df.show
```

1. 保存数据

```scala
scala> var df = spark.read.json("/opt/module/data/input/people.json")
//保存为 parquet 格式
scala> df.write.mode("append").save("/opt/module/data/output")
```

## JSON

Spark SQL 能够自动推测 JSON 数据集的结构，并将它加载为一个 Dataset[Row]. 可以 通过 SparkSession.read.json()去加载 JSON 文件。

注意：Spark 读取的 JSON 文件不是传统的 JSON 文件，每一行都应该是一个 JSON 串。格 式如下：

```json
{"name":"Michael"}
{"name":"Andy"， "age":30}
[{"name":"Justin"， "age":19},{"name":"Justin"， "age":19}]
```

1. 导入隐式换行

`import spark.implicits._`

1. 加载JSON文件

`val path = "/opt/module/spark-local/people.json"`

`val peopleDF = spark.read.json(path)`

1. 创建临时表

`peopleDF.createOrReplaceTempView("people")`

1. 数据查询

```json
val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
teenagerNamesDF.show()
+------+
| name|
+------+
|Justin|
+------+
```

## CSV

Spark SQL 可以配置 CSV 文件的列表信息，读取 CSV 文件,CSV 文件的第一行设置为数据列

```scala
spark.read.format("csv").option("sep", ";").option("inferSchema", "true").option("header", "true").load("data/user.csv")
```

## MySQL

Spark SQL 可以通过 JDBC 从关系型数据库中读取数据的方式创建 DataFrame，通过对 DataFrame 一系列的计算后，还可以将数据再写回关系型数据库中。如果使用 spark-shell 操 作，可在启动 shell 时指定相关的数据库驱动路径或者将相关的数据库驱动放到 spark 的类 路径下。

`bin/spark-shell --jars mysql-connector-java-5.1.27-bin.jar`

IDEA中通过JDBC对MySQL进行操作

1. 导入依赖

```xml
<dependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <version>5.1.27</version>
</dependency>
```

1. 读取数据

```scala
val conf: SparkConf = new
SparkConf().setMaster("local[*]").setAppName("SparkSQL")
//创建 SparkSession 对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
import spark.implicits._
//方式 1：通用的 load 方法读取
spark.read.format("jdbc")
  .option("url", "jdbc:mysql://linux1:3306/spark-sql")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("user", "root")
  .option("password", "123123")
  .option("dbtable", "user")
  .load().show
//方式 2:通用的 load 方法读取 参数另一种形式
spark.read.format("jdbc")
  .options(Map("url"->"jdbc:mysql://linux1:3306/spark-sql?user=root&password=
  123123","dbtable"->"user","driver"->"com.mysql.jdbc.Driver")).load().show
//方式 3:使用 jdbc 方法读取
val props: Properties = new Properties()
props.setProperty("user", "root")
props.setProperty("password", "123123")
val df: DataFrame = spark.read.jdbc("jdbc:mysql://linux1:3306/spark-sql",
                                    "user", props)
df.show
//释放资源
spark.stop()
```

1. 写入数据

```scala
case class User2(name: String, age: Long)
。。。
val conf: SparkConf = new
SparkConf().setMaster("local[*]").setAppName("SparkSQL")
//创建 SparkSession 对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
import spark.implicits._
val rdd: RDD[User2] = spark.sparkContext.makeRDD(List(User2("lisi", 20),
User2("zs", 30)))
val ds: Dataset[User2] = rdd.toDS
//方式 1：通用的方式 format 指定写出类型
ds.write
 .format("jdbc")
 .option("url", "jdbc:mysql://linux1:3306/spark-sql")
 .option("user", "root")
 .option("password", "123123")
 .option("dbtable", "user")
 .mode(SaveMode.Append)
 .save()
//方式 2：通过 jdbc 方法
val props: Properties = new Properties()
props.setProperty("user", "root")
props.setProperty("password", "123123")
ds.write.mode(SaveMode.Append).jdbc("jdbc:mysql://linux1:3306/spark-sql",
"user", props)
//释放资源
spark.stop()
```

## Hive

Apache Hive 是 Hadoop 上的 SQL 引擎，Spark SQL 编译时可以包含 Hive 支持，也 可以不包含。包含 Hive 支持的 Spark SQL 可以支持 Hive 表访问、UDF (用户自定义函数)以及 Hive 查询语言(HiveQL/HQL)等。需要强调的一点是，如果要在 Spark SQL 中包含 Hive 的库，并不需要事先安装 Hive。一般来说，最好还是在编译 Spark SQL 时引入 Hive 支持，这样就可以使用这些特性了。如果你下载的是二进制版本的 Spark，它应该已经在编 译时添加了 Hive 支持。

若要把 Spark SQL 连接到一个部署好的 Hive 上，你必须把 hive-site.xml 复制到 Spark 的配置文件目录中($SPARK_HOME/conf)。即使没有部署好 Hive，Spark SQL 也可以 运行。 需要注意的是，如果你没有部署好 Hive，Spark SQL 会在当前的工作目录中创建出 自己的 Hive 元数据仓库，叫作 metastore_db。此外，如果你尝试使用 HiveQL 中的 CREATE TABLE (并非 CREATE EXTERNAL TABLE)语句来创建表，这些表会被放在你默 认的文件系统中的 /user/hive/warehouse 目录中(如果你的 classpath 中有配好的 hdfs-site.xml，默认的文件系统就是 HDFS，否则就是本地文件系统)。 spark-shell 默认是 Hive 支持的；代码中是默认不支持的，需要手动指定（加一个参数即可）。

1. 内嵌的Hive

如果使用 Spark 内嵌的 Hive, 则什么都不用做, 直接使用即可.

Hive 的元数据存储在 derby 中, 默认仓库地址:$SPARK_HOME/spark-warehouse

```scala
scala> spark.sql("show tables").show
。。。
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
+--------+---------+-----------+
scala> spark.sql("create table aa(id int)")
。。。
scala> spark.sql("show tables").show
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
| default| aa| false|
+--------+---------+-----------+
```

向表加载本地数据

```scala
scala> spark.sql("load data local inpath 'input/ids.txt' into table aa")
。。。
scala> spark.sql("select * from aa").show
+---+
| id|
+---+
| 1|
| 2|
| 3|
| 4|
+---+
```

1. 外部的Hive

如果想连接外部已经部署好的 Hive，需要通过以下几个步骤：

- Spark 要接管 Hive 需要把 hive-site.xml 拷贝到 conf/目录下
- 把 Mysql 的驱动 copy 到 jars/目录下
- 如果访问不到 hdfs，则需要把 core-site.xml 和 hdfs-site.xml 拷贝到 conf/目录下
- 重启 spark-shell

```scala
scala> spark.sql("show tables").show
20/04/25 22:05:14 WARN ObjectStore: Failed to get database global_temp, returning
NoSuchObjectException
+--------+--------------------+-----------+
|database| tableName|isTemporary|
+--------+--------------------+-----------+
| default| emp| false|
| default|hive_hbase_emp_table| false|
| default| relevance_hbase_emp| false|
| default| staff_hive| false|
| default| ttt| false|
| default| user_visit_action| false|
+--
```

1. 运行Spark SQL CLI

Spark SQL CLI 可以很方便的在本地运行 Hive 元数据服务以及从命令行执行查询任务。在 Spark 目录下执行如下命令启动 Spark SQL CLI，直接执行 SQL 语句，类似一 Hive 窗口

`bin/spark-sql`

1. 运行Spark beeline

Spark Thrift Server 是 Spark 社区基于 HiveServer2 实现的一个 Thrift 服务。旨在无缝兼容 HiveServer2。因为 Spark Thrift Server 的接口和协议都和 HiveServer2 完全一致，因此我们部 署好 Spark Thrift Server 后，可以直接使用 hive 的 beeline 访问 Spark Thrift Server 执行相关 语句。Spark Thrift Server 的目的也只是取代 HiveServer2，因此它依旧可以和 Hive Metastore 进行交互，获取到 hive 的元数据。

如果想连接 Thrift Server，需要通过以下几个步骤：

- Spark 要接管 Hive 需要把 hive-site.xml 拷贝到 conf/目录下
- 把 Mysql 的驱动 copy 到 jars/目录下
- 如果访问不到 hdfs，则需要把 core-site.xml 和 hdfs-site.xml 拷贝到 conf/目录下
- 启动 Thrift Server

`sbin/start-thriftserver.sh`

- 使用 beeline 连接 Thrift Server

`bin/beeline -u jdbc:hive2://localhost:10000 -n root`

![](https://cdn.nlark.com/yuque/0/2023/png/33644402/1683621090781-420c0bb9-1990-4fce-a3c6-957add55b8a7.png)

1. 代码操作Hive
2. 导入依赖

```xml
<dependency>
 <groupId>org.apache.spark</groupId>
 <artifactId>spark-hive_2.12</artifactId>
 <version>3.4.0</version>
</dependency>
<dependency>
 <groupId>org.apache.hive</groupId>
 <artifactId>hive-exec</artifactId>
 <version>1.2.1</version>
</dependency>
<dependency>
 <groupId>mysql</groupId>
 <artifactId>mysql-connector-java</artifactId>
 <version>5.1.27</version>
</dependency>
```

1. 将 hive-site.xml 文件拷贝到项目的 resources 目录中，代码实现

```scala
//创建 SparkSession
val spark: SparkSession = SparkSession
.builder()
.enableHiveSupport()
.master("local[*]")
.appName("sql")
.getOrCreate()
```

注意：在开发工具中创建数据库默认是在本地仓库，通过参数修改数据库仓库的地址:

`config("spark.sql.warehouse.dir", "hdfs://localhost:8020/user/hive/warehouse")`

如果在执行操作时，出现如下错误：

![](https://cdn.nlark.com/yuque/0/2023/png/33644402/1683621247802-5d3f8134-94cf-48c1-aaa5-65826c66030c.png)

可以代码最前面增加如下代码解决：

`System.setProperty("HADOOP_USER_NAME", "root")`

此处的 root 改为你们自己的 hadoop 用户名称

SparkSQL数据的加载和保存

# UDF

1. 创建 DataFrame

```scala
scala> val df = spark.read.json("data/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint， username: string]
```

1. 注册UDF

```scala
scala> spark.udf.register("addName",(x:String)=> "Name:"+x)
res9: org.apache.spark.sql.expressions.UserDefinedFunction =
UserDefinedFunction(<function1>,StringType,Some(List(StringType)))
```

1. 创建临时表

```scala
scala> df.createOrReplaceTempView("people")
```

1. 应用UDF

```scala
scala> spark.sql("Select addName(name),age from people").show()
```



# UDAF

强类型的 Dataset 和弱类型的 DataFrame 都提供了相关的聚合函数， 如 count()， countDistinct()，avg()，max()，min()。除此之外，用户可以设定自己的自定义聚合函数。通 过继承 UserDefinedAggregateFunction 来实现用户自定义弱类型聚合函数。从 Spark3.0 版本 后，UserDefinedAggregateFunction 已经不推荐使用了。可以统一采用强类型聚合函数 Aggregator

需求：计算平均工资

一个需求可以采用很多种不同的方法实现需求

1. 实现方式 - RDD

```scala
val conf: SparkConf = new SparkConf().setAppName("app").setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
val res: (Int, Int) = sc.makeRDD(List(("zhangsan", 20), ("lisi", 30), ("wangw", 40))).map {
 case (name, age) => {
 (age, 1)
 }
}.reduce {
 (t1, t2) => {
 (t1._1 + t2._1, t1._2 + t2._2)
 }
}
println(res._1/res._2)
// 关闭连接
sc.stop()
```

1. 实现方式 - 累加器

```scala
class MyAC extends AccumulatorV2[Int,Int]{
 var sum:Int = 0
 var count:Int = 0
 override def isZero: Boolean = {
 return sum ==0 && count == 0
 }
 override def copy(): AccumulatorV2[Int, Int] = {
 val newMyAc = new MyAC
 newMyAc.sum = this.sum
 newMyAc.count = this.count
 newMyAc
 }
 override def reset(): Unit = {
 sum =0
 count = 0
 }
 override def add(v: Int): Unit = {
 sum += v
 count += 1
 }
 override def merge(other: AccumulatorV2[Int, Int]): Unit = {
 other match {
 case o:MyAC=>{
 sum += o.sum
 count += o.count
 }
 case _=>
 }
 }
 override def value: Int = sum/count
}
```

1. 实现方式 - UDAF - 弱类型

```scala
/*
定义类继承 UserDefinedAggregateFunction，并重写其中方法
*/
class MyAveragUDAF extends UserDefinedAggregateFunction {
  // 聚合函数输入参数的数据类型
  def inputSchema: StructType = StructType(Array(StructField("age",IntegerType)))

  // 聚合函数缓冲区中值的数据类型(age,count)
  def bufferSchema: StructType = {
    StructType(Array(StructField("sum",LongType),StructField("count",LongType)))
  }

  // 函数返回值的数据类型
  def dataType: DataType = DoubleType

  // 稳定性：对于相同的输入是否一直返回相同的输出。
  def deterministic: Boolean = true

  // 函数缓冲区初始化
  def initialize(buffer: MutableAggregationBuffer): Unit = {
    // 存年龄的总和
    buffer(0) = 0L
    // 存年龄的个数
    buffer(1) = 0L
  }

  // 更新缓冲区中的数据
  def update(buffer: MutableAggregationBuffer,input: Row): Unit = {
    if (!input.isNullAt(0)) {
      buffer(0) = buffer.getLong(0) + input.getInt(0)
      buffer(1) = buffer.getLong(1) + 1
    }
  }

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

  // 计算最终结果
  def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)
}
。。。
//创建聚合函数
var myAverage = new MyAveragUDAF
//在 spark 中注册聚合函数
spark.udf.register("avgAge",myAverage)
spark.sql("select avgAge(age) from user").show()
```

1. 实现方式 - UDAF - 强类型

```scala
//输入数据类型
case class User01(username:String,age:Long)
//缓存类型
case class AgeBuffer(var sum:Long,var count:Long)
/**
* 定义类继承 org.apache.spark.sql.expressions.Aggregator
* 重写类中的方法
*/
class MyAveragUDAF1 extends Aggregator[User01,AgeBuffer,Double]{
  override def zero: AgeBuffer = {
    AgeBuffer(0L,0L)
  }

  override def reduce(b: AgeBuffer, a: User01): AgeBuffer = {
    b.sum = b.sum + a.age
    b.count = b.count + 1
    b
  }

  override def merge(b1: AgeBuffer, b2: AgeBuffer): AgeBuffer = {
    b1.sum = b1.sum + b2.sum
    b1.count = b1.count + b2.count
    b1
  }

  override def finish(buff: AgeBuffer): Double = {
    buff.sum.toDouble/buff.count
  }

  //DataSet 默认额编解码器，用于序列化，固定写法
  //自定义类型就是 product 自带类型根据类型选择
  override def bufferEncoder: Encoder[AgeBuffer] = {
    Encoders.product
  }

  override def outputEncoder: Encoder[Double] = {
    Encoders.scalaDouble
  }
}
。。。
//封装为 DataSet
val ds: Dataset[User01] = df.as[User01]
//创建聚合函数
var myAgeUdaf1 = new MyAveragUDAF1
//将聚合函数转换为查询的列
val col: TypedColumn[User01, Double] = myAgeUdaf1.toColumn
//查询
ds.select(col).show()
```

Spark3.0 版本可以采用强类型的 Aggregator 方式代替 UserDefinedAggregateFunction

```scala
// TODO 创建 UDAF 函数
val udaf = new MyAvgAgeUDAF
// TODO 注册到 SparkSQL 中
spark.udf.register("avgAge", functions.udaf(udaf))
// TODO 在 SQL 中使用聚合函数
// 定义用户的自定义聚合函数
spark.sql("select avgAge(age) from user").show
// **************************************************
case class Buff( var sum:Long, var cnt:Long )
// totalage, count
class MyAvgAgeUDAF extends Aggregator[Long, Buff, Double]{
  override def zero: Buff = Buff(0,0)

  override def reduce(b: Buff, a: Long): Buff = {
    b.sum += a
    b.cnt += 1
    b
  }

  override def merge(b1: Buff, b2: Buff): Buff = {
    b1.sum += b2.sum
    b1.cnt += b2.cnt
    b1
  }

  override def finish(reduction: Buff): Double = {
    reduction.sum.toDouble/reduction.cnt
  }

  override def bufferEncoder: Encoder[Buff] = Encoders.product

  override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}
```

UDF/UDAF

# 实现原理

累加器用来把 Executor 端变量信息聚合到 Driver 端。在 Driver 程序中定义的变量，在 Executor 端的每个 Task 都会得到这个变量的一份新的副本，每个 task 更新这些副本的值后， 传回 Driver 端进行 merge。

# 基础编程

### 系统累加器

```scala
val rdd = sc.makeRDD(List(1,2,3,4,5))
// 声明累加器
var sum = sc.longAccumulator("sum");
rdd.foreach(
 num => {
 // 使用累加器
 sum.add(num)
 }
)
// 获取累加器的值
println("sum = " + sum.value)
```



### 自定义累加器

```scala
// 自定义累加器
// 1. 继承 AccumulatorV2，并设定泛型
// 2. 重写累加器的抽象方法
class WordCountAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]]{
  var map : mutable.Map[String, Long] = mutable.Map()
  // 累加器是否为初始状态
  override def isZero: Boolean = {
   map.isEmpty
  }

  // 复制累加器
  override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = {
   new WordCountAccumulator
  }

  // 重置累加器
  override def reset(): Unit = {
   map.clear()
  }

  // 向累加器中增加数据 (In)
  override def add(word: String): Unit = {
   // 查询 map 中是否存在相同的单词
   // 如果有相同的单词，那么单词的数量加 1
   // 如果没有相同的单词，那么在 map 中增加这个单词
   map(word) = map.getOrElse(word, 0L) + 1L
  }

  // 合并累加器
  override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]):
  Unit = {
   val map1 = map
   val map2 = other.value
   // 两个 Map 的合并
   map = map1.foldLeft(map2)(
   ( innerMap, kv ) => {
   innerMap(kv._1) = innerMap.getOrElse(kv._1, 0L) + kv._2
   innerMap
   }
   )
  }
  // 返回累加器的结果 （Out）
  override def value: mutable.Map[String, Long] = map
}
```

基本命令

1. ls - 列出目录内容

2. cd - 切换目录

通用的加载和保存方式

UDF

实现原理

基础编程

系统累加器

1. `ls` - 列出目录内容

2. `cd` - 切换目录