# addition-homework1

**Repository Path**: zero-night-01/addition-homework1

## Basic Information

- **Project Name**: addition-homework1
- **Description**: 极客时间加餐-作业1，SparkSQL
- **Primary Language**: Scala
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2022-05-22
- **Last Updated**: 2022-05-30

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# addition-homework1

#### 介绍
极客时间加餐-作业1，SparkSQL

#### 使用说明

实现的功能：

- 基本功能，将表的数据合并成指定的文件数
- 不添加 fileNum，合并成 128MB 大小

实现过程

在 `SqlBase.g4` 中添加如下内容

statement 下

```
| COMPACT TABLE target=tableIdentifier partitionSpec?
    (INTO fileNum=INTEGER_VALUE FILES)?                         #compactTable
```

ansiNonReserved 和 nonReserved 下

```
| FILES
```

keyword list 下

```
FILES: 'FILES'
```

运行 

```shell
build/mvn antlr4:antlr4 -f pom.xml
```

编写代码

`SparkSqlParser.scala`中添加

```scala
override def visitCompactTable(ctx: CompactTableContext): LogicalPlan = withOrigin(ctx) {
    val tableIdentifier = visitTableIdentifier(ctx.tableIdentifier())
    val fileNum: String = ctx.fileNum match {
      case null => ""
      case t: Token => t.getText
    }
    CompactTableCommand(tableIdentifier, fileNum)
  }
```

实现 CompactTableCommand

```scala
case class CompactTableCommand(table: TableIdentifier,
                               fileNum: String) extends LeafRunnableCommand {
  override val output: Seq[Attribute] =
    Seq(AttributeReference("result", StringType, nullable = false)())

  override def run(sparkSession: SparkSession): Seq[Row] = {
    val dataFrame = sparkSession.table(tableIdent = table)
    val oldPartitions = dataFrame.rdd.getNumPartitions
    val m128: Long = 128L * 1024L * 1024L
    val nums: Int = fileNum match {
      case "" => Math.max(
        (dataFrame.queryExecution.optimizedPlan.stats.sizeInBytes.toLong / m128).toInt, 1)
      case i: String => i.toInt
    }
    // write to tmp table
    val tmpTable = table.identifier + "_tmp"
    dataFrame.repartition(nums).write.mode(SaveMode.Overwrite).saveAsTable(tmpTable)
    sparkSession.table(tmpTable).write.mode(SaveMode.Overwrite).saveAsTable(table.identifier)
    sparkSession.sql(s"drop table if exists $tmpTable")
    Seq(Row(s"${table.database}.${table.table} old partitions: $oldPartitions, " +
      s"new partitions: $nums"))
  }
}
```

思路：

1. 在 `visitCompactTable` 中判断是否有`fileNum`参数，以此在`CompactTableCommand`中决定是否使用 `128M` 大小的文件来计算新的文件数

2. 通过`repartition`方法减少分区数，以减少文件数，先写入临时表以防止异常时影响原表

准备工作

```sql
CREATE TABLE tmp (
    id int,
    t string
);
INSERT INTO TABLE tmp VALUES (1, "test1");
INSERT INTO TABLE tmp VALUES (2, "test2");
INSERT INTO TABLE tmp VALUES (2, "test2");
INSERT INTO TABLE tmp VALUES (2, "test2");
INSERT INTO TABLE tmp VALUES (2, "test2");
INSERT INTO TABLE tmp VALUES (3, "test3");
INSERT INTO TABLE tmp VALUES (3, "test3");
INSERT INTO TABLE tmp VALUES (4, "test4");
INSERT INTO TABLE tmp VALUES (5, "test5");
```

运行命令

```sql
COMPACT TABLE tmp INTO 2 FILES;
```

运行结果

![image-20220522195203193](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220522195203193.png)

验证不带文件数的情况

![image-20220522223114788](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220522223114788.png)