使用Flink SQL通过Paimon REST访问DLF Catalog-数据湖构建-阿里云

如何在阿里云实时计算Flink版上通过Flink SQL以Paimon REST访问DLF Catalog。

前提条件

已创建Flink全托管工作空间。如未创建，详情请参见开通实时计算Flink版。
请确保Flink工作空间与DLF位于同一地域下，且添加Flink所在VPC为白名单，请参见配置VPC白名单。

使用限制

仅实时计算引擎VVR 11.1.0及以上版本支持对接DLF Catalog。

创建DLF Catalog

详情请参见DLF 快速入门。

在Flink上添加Catalog

说明

这里仅用于建立与DLF Catalog的映射连接，创建或删除Catalog仅影响映射关系，不会对DLF中的实际数据产生影响。

登录实时计算控制台。
单击Flink全托管工作空间列表中的工作空间名称，进入对应的工作空间。

可以通过UI或者SQL方式来添加Catalog。

UI方式

在左侧导航栏，单击数据管理。
在Catalog列表页，单击创建Catalog。
选择Apache Paimon，单击下一步。
metastore 选择 dlf，catalog name 选择需要关联版本的DLF Catalog，单击确定，即可创建Catalog。

SQL命令

在数据查询文本编辑区域，输入以下SQL并执行，即可创建Catalog。

CREATE CATALOG `flink_catalog_name` 
WITH (
  'type' = 'paimon',
  'metastore' = 'rest',
  'token.provider' = 'dlf',
  'uri' = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com',
  'warehouse' = 'dlf_test'
);

参数说明如下：

参数	描述	是否必填	示例
type	Catalog类型，固定为paimon。	是	paimon
metastore	metastore类型，固定为rest。	是	rest
token.provider	Token提供方，固定为dlf。	是	dlf
uri	访问DLF Rest Catalog Server的URI，格式为`http://[region-id]-vpc.dlf.aliyuncs.com`。详见服务接入点中的Region ID。	是	http://cn-hangzhou-vpc.dlf.aliyuncs.com
warehouse	DLF Catalog名称。	是	dlf_test

常见报错

Http NullPointerException

报错栈示例：

Caused by: java.lang.NullPointerException
	at org.apache.paimon.rest.HttpClient.getRequestUrl(HttpClient.java:180)
	at org.apache.paimon.rest.HttpClient.get(HttpClient.java:108)
	at org.apache.paimon.rest.RESTCatalog.<init>(RESTCatalog.java:159)
	at org.apache.paimon.rest.RESTCatalog.<init>(RESTCatalog.java:141)
	at org.apache.paimon.rest.RESTCatalogFactory.create(RESTCatalogFactory.java:36)
	at org.apache.paimon.catalog.CatalogFactory.createUnwrappedCatalog(CatalogFactory.java:84)
	at org.apache.paimon.catalog.CatalogFactory.createCatalog(CatalogFactory.java:71)
	at org.apache.paimon.flink.VvrCatalogFactory.createCatalog(VvrCatalogFactory.java:67)
	at org.apache.paimon.flink.VvrCatalogFactory.createCatalog(VvrCatalogFactory.java:37)
	at org.apache.flink.table.gateway.vvr.module.delegation.CatalogFactoryWrapper.createCatalog(CatalogFactoryWrapper.java:91)
	at org.apache.flink.table.factories.FactoryUtil.createCatalog(FactoryUtil.java:572)
	... 23 more

解决方法：
检查catalog配置项中uri的值，应该以http://开头。

ForbiddenException: Forbidden: You are not authorized to do this operation

报错栈示例：

Caused by: org.apache.paimon.rest.exceptions.ForbiddenException: Forbidden: You are not authorized to do this operation. Action: dlf:GetConfig. AccessDeniedDetail : {"AuthAction":"dlf:GetConfig","AuthPrincipalType":"AssumedRoleUser","AuthPrincipalOwnerId":"xxxxx","AuthPrincipalDisplayName":"aliyunstreamasidefaultrole:xxxxx","PolicyType":"AccountLevelIdentityBasedPolicy","NoPermissionType":"ImplicitDeny"} requestId:xxxxx
	at org.apache.paimon.rest.DefaultErrorHandler.accept(DefaultErrorHandler.java:59)
	at org.apache.paimon.rest.DefaultErrorHandler.accept(DefaultErrorHandler.java:35)
	at org.apache.paimon.rest.HttpClient.exec(HttpClient.java:205)
	at org.apache.paimon.rest.HttpClient.get(HttpClient.java:112)
	at org.apache.paimon.rest.RESTCatalog.<init>(RESTCatalog.java:159)
	at org.apache.paimon.rest.RESTCatalog.<init>(RESTCatalog.java:141)
	at org.apache.paimon.rest.RESTCatalogFactory.create(RESTCatalogFactory.java:36

解决方法：
授予AliyunStreamAsiDefaultRole角色AliyunDLFFullAccess权限（Flink产品化集成后可以省略该步骤）。
1. 使用阿里云账号或RAM管理员登录RAM控制台。
2. 单击身份管理 > 角色，查询AliyunStreamAsiDefaultRole角色。
3. 单击操作列的新增授权，进入新增授权页面。
4. 在权限策略中，查询并勾选AliyunDLFFullAccess，单击确认新增授权。