原文链接
作者：马骏杰
问题
SparkSQL是Spark的一个子模块，主要功能是用于处理结构化数据，目前在大数据OLAP领域已经有了广泛的应用。Iceberg作为一个通用的表格式，也已经在数据湖的解决方案中逐渐展现了它的优势。
那该如何将这2者相结合，使得应用SparkSQL + Iceberg可以和SparkSQL + Hive一样方便，如，基于SQL直接访问数据或进行DDL操作：
select c1 from iceberg_db.t;
drop table iceberg_db.t;
SparkSQL 基本原理
先来看下SparkSQL处理SQL的基本流程：
如上图所示，在提交SQL后，spark内部会经历语法解析生成逻辑计划，解析逻辑计划，优化逻辑计划，生成执行计划，执行。在解析逻辑计划的过程中，引入了catalog，它的作用是来判断SQL中引用的数据库，表，列，函数等是否存在。
在Spark + Hive的解决方案中，基于ExternalCatalog接口，实现了HiveExternalCatalog，该类中的Hive客户端和Hive的metastore进行交互，从而能解析SQL中的库表列是否存在，并能基于Hive客户端进行Hive表的DDL操作，比如create table， drop table等。
Multiple Catalog解析
那Spark + Iceberg是否只需要实现ExternalCatalog接口，就能基于SQL直接访问数据或进行DDL操作吗？答案是肯定的，但是，由于解析SQL过程中只能支持一种catalog，如果要实现Hive table joion Iceberg table该怎么办，如：
select *
from iceberg_db.t1
join hive_db.t2
on t1.k1 = t2.k1;
为了更为通用的解决这类问题，在Spark 3.0 preview版本中引入了multiple catalog功能，该功能对于catalog做了如下变化：
- 增加了CatalogPlugin接口，所有新增数据源的自定义catalog都要实现该接口。该接口提供了统一的初始化方法，CatalogManager(下文会提到)在创建CatalogPlugin实例后会进行用户实现的初始化逻辑。在使用过程中，需要增加配置
spark.sql.catalog.<catalog-name>=<YourCatalogClass>
这里设置为需要的CatalogName，设置为具体的实现类，如，
spark.sql.catalog.iceberg_catalog = org.apache.iceberg.spark.SparkCatalog
接口定义如下：
public interface CatalogPlugin {
void initialize(String name, CaseInsensitiveStringMap options);
String name();
}
- 增加了TableCatalog的接口，该接口继承自CatalogPlugin，定义了相关的方法用来解析SQL中的元数据，如，tableExists，还定义了一系列方法进行DDL操作，如，createTable，alterTable，dropTable，接口定义如下，
public interface TableCatalog extends CatalogPlugin {
Identifier[] listTables(String[] namespace) throws NoSuchNamespaceException;
Table loadTable(Identifier ident) throws NoSuchTableException;
default void invalidateTable(Identifier ident) {
}
default boolean tableExists(Identifier ident) {
try {
return loadTable(ident) != null;
} catch (NoSuchTableException e) {
return false;
}
}
Table createTable(
Identifier ident,
StructType schema,
Transform[] partitions,
Map<String, String> properties) throws TableAlreadyExistsException, NoSuchNamespaceException;
Table alterTable(
Identifier ident,
TableChange... changes) throws NoSuchTableException;
boolean dropTable(Identifier ident);
void renameTable(Identifier oldIdent, Identifier newIdent)
throws NoSuchTableException, TableAlreadyExistsException;
}
- 增加了CatalogManager，该类的作用是在解析逻辑计划过程中，基于catalogName，返回匹配的CatalogPlugin实现。由于CatalogManager的增加，解析SQL时和catalog的交互也发生了变化：
在解析过程中，根据catalogName从CatalogManager获取具体的CatalogPlugin实现，V2SessionCatalog是为了兼容之前的catalog的实现机制，而CustomerCatalog是自定义的CatalogPlugin实现。同时，CatalogManager还会管理当前的Catalog/Namespace，相关方法如下：
def currentNamespace: Array[String]
def setCurrentNamespace(namespace: Array[String]): Unit
def currentCatalog: CatalogPlugin
def setCurrentCatalog(catalogName: String): Unit
- 命名结构变更为.*，对于表名，原本只支持2层的命名结构，databaseName.tableName，但是在业界流行的数据库中（如MySQL，PostgreSQL），已经支持3层的命名结构，database.schema.table。而在multiple catalog实现过程中，引入了Namespace概念，使得SparkSQL能支持多层命名结构，如，catalog.ns1.ns2.table
由于引入了catalog和namespace概念，SparkSQL还增加相关命令支持catalog/namespace的管理，如，
CREATE/DROP/SHOW NAMESPACES
USE <catalogName>.<namespaceName>
除了multiple catalog以外，SparkSQL DatasourceV2还重构生成了SupportsRead/SupportsWrite等接口，用来支持数据源的各类操作，由于篇幅有限，就不在本文中具体展开。
基于 Spark 3.0 preview使用Iceberg + SparkSQL
在Spark DatasourceV2增加了multiple catalog等功能后，回到我们想要查询的SQL，实现步骤如下：
1.在Iceberg侧对CatalogPlugin/TableCatalog/SupportsRead等接口进行实现，实现类名如: org.apache.iceberg.spark.SparkCatalog
2.在spark的配置文件中设置：
spark.sql.catalog.iceberg_catalog = org.apache.iceberg.spark.SparkCatalog
3.基于配置的catalogName，调整SQL如下，就可以进行基于SQL的跨数据源查询了。
select *
from iceberg_catalog.ns1.t1
join hive_db.t2 on t1.k1 = t2.k1;
4.除了跨数据源数据分析以外，现在还可以对Iceberg的表进行DDL操作了，如，
create table iceberg_catalog.t1 ......
drop table iceberg_catalog.t1
总结
Spark 3.0 preview在DatasourceV2的功能方面较Spark2.4做了比较大的改动，Multiple Catalog作为比较重要的新增功能，使得新的数据源能很便捷的和SparkSQL进行整合，提供元数据相关服务。除了Multiple Catalog以外，还增加了诸如SupportsRead，SupportsWrite，SupportsPushDownFilters等一系列接口增强对数据源的整合。Iceberg作为新兴的表格式，能很好的利用DatasourceV2的新功能，结合SparkSQL构建数据湖解决方案。目前Iceberg开源代码还未针对新的DatasourceV2特性进行更新，在我们内部项目中已经对这块整合进行了相关实践，并计划贡献给社区，使得基于Iceberg的数据湖解决方案能更完善。
阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区，定期推送精彩案例，技术专家直播，问答区近万人Spark技术同学在线提问答疑，只为营造纯粹的Spark氛围，欢迎钉钉扫码加入！
对开源大数据和感兴趣的同学可以加小编微信（下图二维码，备注“进群”）进入技术交流微信群。
Spark技术交流社区公众号，微信扫一扫关注
