当内置的连接器无法满足需求时,您可以使用自定义连接器。本文为您介绍如何上传、使用和更新自定义连接器。
注意事项
您需要根据社区定义的连接器标准开发您自定义的连接器,才能正确地被实时计算Flink版识别并使用,因此您需要明确连接器Meta文件和声明Factory类。自定义连接器开发详情请参见User-defined Sources & Sinks。
同种类型的连接器仅可以上传一次。
为了避免JAR包依赖冲突,您还需要注意以下几点:
Flink镜像和Pom依赖的Flink版本请保持一致。
请不要上传Runtime层的JAR包,即在依赖中添加
<scope>provided</scope>
。其他第三方依赖请采用Shade方式打包,Shade打包详情参见Apache Maven Shade Plugin。
上传与使用自定义连接器
进入自定义连接器注册入口。
登录实时计算控制台。
单击目标工作空间操作列下的控制台。
在左侧导航栏,单击数据连接。
注册自定义连接器。
在数据连接页面,单击创建自定义连接器。
上传自定义连接器JAR文件。
您可以通过以下任何一种方式上传自定义连接器JAR文件:
上传文件:单击选择文件后,选择您的目标连接器JAR文件。
外部URL:当需要使用其他服务上存在的JAR文件时,可以使用外部URL功能获取JAR文件。例如,https://ossbucket/artifacts/namespaces/flink-default/flink-jobs-1.0-SNAPSHOT.jar
说明在使用其他服务的HTTP路径时,需要该服务和Flink工作空间在同一VPC下,或者先打通Flink工作空间和目标服务的公网网络后,使用公网地址访问,详情请参见Flink全托管集群如何访问公网?。
上传完成后,单击下一步。
系统会对您上传的自定义连接器内容进行解析。如果解析成功,您可以继续下一步。如果解析失败,请确认您上传的自定义连接器代码是否符合Flink社区标准。
单击完成。
创建完成的自定义连接器会出现在连接器列表中。
在作业DDL中使用连接器。
作业开发详情请参见SQL作业开发。
说明WITH参数中的连接器取值为您自定义连接器JAR包中DynamicTableFactory的identifier参数取值,其他WITH参数及含义详情由您开发的自定义连接器决定。
更新自定义连接器
更新自定义连接器后,使用该连接器的运行中的作业不受影响。重新启动作业后,会使用更新的连接器。
进入自定义连接器更新入口。
登录实时计算控制台。
单击目标工作空间操作列下的控制台。
在左侧导航栏,单击数据连接。
在自定义连接器页签,单击目标自定义连接器名称右侧的编辑。
上传自定义连接器JAR文件。
您可以通过以下任何一种方式上传自定义连接器JAR文件:
上传文件:单击选择文件后,选择您的目标连接器JAR文件。
外部URL:输入外部URL地址。例如,https://ossbucket/artifacts/namespaces/flink-default/flink-jobs-1.0-SNAPSHOT.jar
说明如果外部URL是OSS Bucket地址,则自定义连接器文件必须位于sql-artifacts/namespaces/{namespace}目录下。
如果您的连接器文件存放在非OSS的其他服务上,且和Flink工作空间不在同一个VPC,则此处需要填写为公网服务地址。需确认已打通Flink和目标服务的公网网络连接,详情请参见Flink全托管如何访问公网?。
上传完成后,单击下一步。
系统会对您上传的自定义连接器内容进行解析。如果解析成功,您可以继续下一步。如果解析失败,请确认您上传的自定义连接器代码是否符合Flink社区标准。
单击完成。
相关文档
Flink内置了丰富的连接器,支持的内置连接器详情请参见支持的连接器。
使用社区版CDC连接器,详情请参见社区版CDC连接器使用方法。
Flink提供了元数据管理,您可以通过创建Catalog来管理和访问相关的元数据,详情请参见管理元数据。
- 本页导读 (1)