# 在 Spark 中访问 OushuDB 元数据

OushuDB 提供了 OushuDB Catalog 插件为 Spark 提供元数据，并允许 Spark 根据 OushuDB 元数据来读取数据。

## 安装

### 复制 OushuDB Catalog 到 Spark 下

将 OushuDB RPM 包中的 OushuDB Catalog JAR 包拷贝到 `${SPARK_HOME}/jars` 目录下。

OushuDB Catalog JAR 包命名形式为 `oushudb-catalog-<oushudb_catalog_version>-spark<spark_version>_bundle.jar`, 其中 `<oushudb_catalog_version>` 为 OushuDB Catalog 的版本号， `<spark_version>` 为对应的 Spark 版本号。

`${SPARK_HOME}` 为 Spark 的安装目录。

若用户不希望直接复制 OushuDB Catalog JAR 包到 Spark 目录下， 也可以通过在使用 Spark 时指定 `--jars=<path_to_oushudb_catalog_jar>` 的方式来显示地提供。

### 在 OushuDB 中创建 spark_toolkit 扩展

为了使用 OushuDB Catalog 的 Spark 功能， 用户需要登录到 OushuDB 中创建 `spark_toolkit` 扩展。 `spark_toolkit` 扩展随 OushuDB 安装包发布，但缺省不会安装，用户可以通过以下命令完成：

```sql
postgres=# CREATE EXTENSION spark_toolkit;
CREATE EXTENSION
```

## 使用说明

在 Spark 中加载 OushuDB Catalog 只需要按照一般的扩展插件的加载方式进行。用户可以通过指定 `spark.sql.extensions=com.oushu.catalog.oushudb.OushuDBSparkSessionExtension` 来显示的加载。

此外运行 OushuDB Catalog 还需要指定和 OushuDB 相关的如下配置:

```
spark.sql.catalog.oushu_catalog=org.apache.spark.sql.oushudb.catalog.OushuDBCatalog
spark.sql.catalog.oushu_catalog.oushu.db.address=<hostname>:<port>
spark.sql.catalog.oushu_catalog.oushu.db.database=<database_name>
```

其中 `spark.sql.catalog.oushu_catalog=org.apache.spark.sql.oushudb.catalog.OushuDBCatalog` 指明了 OushuDB Catalog 使用的是 Spark 元数据组件， `spark.sql.catalog.oushu_catalog.oushu.db.address=<hostname>:<port>` 配置了 OushuDB 服务的监听地址，而 `spark.sql.catalog.oushu_catalog.oushu.db.database=<database_name>` 配置了 OushuDB Catalog 连接的数据库名称。

当 OushuDB Catalog 加载后，用户可以通过 `oushu_catalog` 来访问 OushuDB 的元数据。

## 数据类型

OushuDB Catalog 目前仅支持《数据类型对照表》中的数据类型。

### 数据类型对照表
| OushuDB数据类型                      | Spark数据类型 |
|--------------------------------------|---------------|
| boolean/bool                         | BooleanType   |
| smallint/smallserial/int2            | ShortType     |
| integer/serial/int/int4              | IntegerType   |
| bigint/bigserial/oid/int8/long       | LongType      |
| real/float/float4                    | FloatType     |
| double/doubleprecision/float8/money  | DoubleType    |
| json/string/text                     | StringType    |
| bit/bitvarying/bytea/binary          | BinaryType    |
| date                                 | DateType      |
| timestamp with time zone/timestamptz | TimestampType |
| numeric/decimal                      | DecimalType   |
| character varying/varchar            | VarcharType   |
| character/char/bpchar                | CharType      |
| array                                | ArrayType     |

注意:
* OushuDB 以 Hudi Parquet 格式存储 `timestamptz` 类型时，会将数据归一化到 UTC 时间，此行为与 Hudi Parquet 社区版本行为一致，Spark 会自动将 UTC 时间偏移到当前时区。
* OushuDB 以 Hudi ORC 格式存储 `timestamptz` 类型时，会将数据归一化到 UTC 时间，但 Hudi ORC 社区版本存储数据时不会归一化时间，为了能在 Spark 中正确读取数据，需要使用 OushuDB Catalog 插件自动将 UTC 时间偏移到当前时区后传递给 Spark。

## 示例

### 通过 spark-sql 获取 OushuDB 中的表
```
spark-sql \
    --conf spark.sql.extensions=com.oushu.catalog.oushudb.OushuDBSparkSessionExtension \
    --conf spark.sql.catalog.oushu_catalog=org.apache.spark.sql.oushudb.catalog.OushuDBCatalog \
    --conf spark.sql.catalog.oushu_catalog.oushu.db.address=localhost:7000 \
    --conf spark.sql.catalog.oushu_catalog.oushu.db.database=postgres \
    -e "show tables in oushu_catalog"
```

### 通过 spark-sql 查询 OushuDB 中的表
```
spark-sql \
    --conf spark.sql.extensions=com.oushu.catalog.oushudb.OushuDBSparkSessionExtension \
    --conf spark.sql.catalog.oushu_catalog=org.apache.spark.sql.oushudb.catalog.OushuDBCatalog \
    --conf spark.sql.catalog.oushu_catalog.oushu.db.address=localhost:7000 \
    --conf spark.sql.catalog.oushu_catalog.oushu.db.database=postgres \
    -e "select * from oushu_catalog.schema_name.table_name"
```

## 注意事项

* OushuDB Catalog 仅适配 Spark 3.1, 其他 Spark 版本未经测试可能存在问题。
* OushuDB Catalog 目前仅支持读取功能，不支持 DDL 和 DML。
* OushuDB Catalog 目前仅支持 Magam 表的元数据。
* OushuDB Catalog 目前仅支持 Hudi Parquet 和 Hudi ORC 存储格式。
* OushuDB Catalog 无法保证 Spark 事务与 OushuDB 事务的并发正确，为了避免错误，请关闭 OushuDB 的 autocompact，autoarchive等后台任务。
* 当 Spark 读取 OushuDB 中启用压缩算法的表时，Spark 依赖的 HADOOP 必须提供对应的压缩功能, 否则 Spark 将会抛出类似如下的错误 `java.lang.RuntimeException: native zStandard library not available: this version of libhadoop was built without zstd support.`。