引擎对接 - Spark-实时计算 Flink版(Flink)-阿里云帮助中心

本文为您介绍如何在阿里云 EMR Serverless Spark 中通过 Fluss Spark Connector 读写 Fluss 数据。

前提条件

已开通流存储 Apache Fluss 版，并在 Fluss 控制台目标集群的概览页面获取 bootstrap.servers 地址与内置账号信息。
已创建 EMR Serverless Spark 工作空间。详见创建工作空间。
已连通 EMR Serverless Spark 工作空间与 Fluss 集群所在VPC。详见EMR Serverless Spark与其他VPC间网络互通。

使用限制

EMR Serverless Spark 引擎版本 esr-4.9.0 及以上内置 Fluss Spark Connector，支持读写所有类型的 Fluss 表。
通过 Maven 下载的指定版本 Connector 当前仅支持读写 Fluss 日志表。
低于 esr-4.9.0 的引擎版本不内置 Connector，需引用指定版本 JAR 包，详见使用指定版本 Connector。

注册 Fluss Catalog

EMR Serverless Spark 自 esr-4.9.0 起内置 Fluss Spark Connector，在创建批任务或会话时通过 Spark 配置注册 Fluss Catalog 即可使用。

登录 EMR Serverless Spark 控制台，进入目标工作空间。
创建批任务或会话。
- 批任务：在左侧导航栏单击数据开发，新建 SQL 类型批任务。
- SQL 会话/Notebook 会话：在左侧导航栏单击会话管理，单击创建 SQL 会话或创建 Notebook 会话。
配置以下参数。
- 引擎版本：选择 esr-4.9.0 及以上版本。
- 网络连接：选择已创建并打通 Fluss 集群 VPC 的网络连接。
- 资源队列：选择对应队列。

在自定义配置 > Spark 配置中追加以下配置，注册 Fluss Catalog。

spark.sql.catalog.fluss_catalog org.apache.fluss.spark.SparkCatalog
spark.sql.catalog.fluss_catalog.bootstrap.servers <bootstrap_servers>
spark.sql.extensions org.apache.fluss.spark.FlussSparkSessionExtensions
spark.sql.catalog.fluss_catalog.client.security.protocol SASL
spark.sql.catalog.fluss_catalog.client.security.sasl.mechanism PLAIN
spark.sql.catalog.fluss_catalog.client.security.sasl.username <username>
spark.sql.catalog.fluss_catalog.client.security.sasl.password <password>

各配置项说明如下。

配置项	说明	是否必填	备注
`spark.sql.catalog.fluss_catalog`	Catalog 实现类。	是	固定值为 `org.apache.fluss.spark.SparkCatalog`。`fluss_catalog` 为自定义的 Catalog 名称，可按需修改，后续 SQL 中以此名称引用。
`spark.sql.catalog.fluss_catalog.bootstrap.servers`	连接 Fluss 集群的 bootstrap server 地址。	是	服务地址。
`spark.sql.extensions`	Fluss Spark SQL 扩展类。	是	固定值为 `org.apache.fluss.spark.FlussSparkSessionExtensions`。
`spark.sql.catalog.fluss_catalog.client.security.protocol`	鉴权框架。	是	固定值为 `SASL`。
`spark.sql.catalog.fluss_catalog.client.security.sasl.mechanism`	鉴权机制。	是	固定值为 `PLAIN`。
`spark.sql.catalog.fluss_catalog.client.security.sasl.username`	内置用户名。	是	Fluss 实例内置用户名。
`spark.sql.catalog.fluss_catalog.client.security.sasl.password`	内置密码。	是	Fluss 实例内置密码。

单击创建或运行，完成 Catalog 注册。

管理库表

完成 Catalog 注册后，在批任务或 SQL 会话/Notebook 会话的 SparkSQL 脚本中执行以下语句，管理 Fluss 的库表。

-- 查看 Catalog 下所有数据库
SHOW DATABASES IN fluss_catalog;
-- 创建数据库
CREATE DATABASE IF NOT EXISTS fluss_catalog.testdb;
-- 查看数据库下的所有表
SHOW TABLES IN fluss_catalog.testdb;
-- 创建表
CREATE TABLE IF NOT EXISTS fluss_catalog.testdb.test (id INT, name STRING);

写入数据

使用 INSERT INTO 向 Fluss 表写入数据。

INSERT INTO fluss_catalog.testdb.test VALUES(1, 'Bob');

读取数据

批读取

默认以批模式读取 Fluss 表的全量数据。

SELECT * FROM fluss_catalog.testdb.test;

加速读取

加速读取仅扫描主键表的快照数据，跳过日志合并步骤，适用于对一致性要求不高、对延迟敏感的查询场景。

SET spark.sql.fluss.read.optimized=true;
SELECT * FROM fluss_catalog.testdb.test;

湖流一体读取（Union Read）

Fluss 表开启湖流一体后，查询会自动合并实时 Fluss 数据与已归档至数据湖（Paimon）的历史数据，无需在 SQL 层手动拼接。

为表开启湖流一体存储。

ALTER TABLE fluss_catalog.testdb.test SET TBLPROPERTIES ('table.datalake.enabled' = 'true');

直接查询表即可获得湖流合并结果。
```
SELECT * FROM fluss_catalog.testdb.test;
```

使用指定版本 Connector

当内置 Connector 无法满足需求，或引擎版本低于 esr-4.9.0 时，可手动引用指定版本的 Fluss Spark Connector。

下载 Fluss Spark Connector JAR。

curl -fLO "https://repo1.maven.org/maven2/org/apache/fluss/fluss-spark-3.5_2.12/0.9.1-incubating/fluss-spark-3.5_2.12-0.9.1-incubating.jar"

将 JAR 包上传至 OSS。
在创建批任务或会话时，Spark 配置中除注册 Catalog 的配置外，再追加以下两项，排除内置模块并加载自定义 JAR。
```
spark.emr.serverless.excludedModules fluss
spark.emr.serverless.user.defined.jars oss://<bucket>/path/fluss-spark-3.5_2.12-0.9.1-incubating.jar  
```