开源大数据EMR
2020-04-23
796浏览量
原文链接
作者:马骏杰
SparkSQL是Spark的一个子模块,主要功能是用于处理结构化数据,目前在大数据OLAP领域已经有了广泛的应用。Iceberg作为一个通用的表格式,也已经在数据湖的解决方案中逐渐展现了它的优势。
那该如何将这2者相结合,使得应用SparkSQL + Iceberg可以和SparkSQL + Hive一样方便,如,基于SQL直接访问数据或进行DDL操作:
select c1 from iceberg_db.t;
drop table iceberg_db.t;
先来看下SparkSQL处理SQL的基本流程:
如上图所示,在提交SQL后,spark内部会经历语法解析生成逻辑计划,解析逻辑计划,优化逻辑计划,生成执行计划,执行。在解析逻辑计划的过程中,引入了catalog,它的作用是来判断SQL中引用的数据库,表,列,函数等是否存在。
在Spark + Hive的解决方案中,基于ExternalCatalog接口,实现了HiveExternalCatalog,该类中的Hive客户端和Hive的metastore进行交互,从而能解析SQL中的库表列是否存在,并能基于Hive客户端进行Hive表的DDL操作,比如create table, drop table等。
那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做了如下变化:
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();
}
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;
}
在解析过程中,根据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
CREATE/DROP/SHOW NAMESPACES
USE <catalogName>.<namespaceName>
除了multiple catalog以外,SparkSQL DatasourceV2还重构生成了SupportsRead/SupportsWrite等接口,用来支持数据源的各类操作,由于篇幅有限,就不在本文中具体展开。
在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技术交流社区公众号,微信扫一扫关注
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。
阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区,定期推送精彩案例,问答区数个Spark技术同学每日在线答疑,只为营造纯粹的Spark氛围,欢迎加入!邀请你加入钉钉群聊Apache Spark中国技术交流社区,点击进入查看详情 https://qr.dingtalk.com/action/joingroup?code=v1,k1,X7S/0/QcrLMkK7QZ5sw2oTvoYW49u0g5dvGu7PW+sm4=&_dt_no_comment=1&origin=11