一般来说,在 Flink SQL Client 中使用各种 Connector 只需要该 Connector 及其依赖 Jar 包部署到 ${FLINK_HOME}/lib 下即可。但是对于某些特定的平台,如果 AWS EMR、Cloudera CDP 等产品会有所不同,主要是它们中的某些 Jar 包可能被改写过,例如和 Hive Metastore 的交互,AWS EMR 就有另外一套 Metatstore:Glue Data Catalog,所以接口也做了相应的,所以,简单的复制开源的 Jar 包可能会有问题,最好做法还是从该平台/产品的集群上拷贝本地的 Jar 包。

以下脚本,以 EMR 6.15 ( Flink 1.17.1)为例,展示了各类常用 Connector 的安装方法,有的是直接下载自开源社区,有的则是从 EMR 集群本地找到相应 Jar 包安装的。脚本在 EMR 6.15 上全部测试通过,如果在其他版本的 EMR 或 Flink 上安装,请注意修改版本号。

FLINK_VERSION="1.17.1"FLINK_MAJOR_VERSION="1.17"HUDI_VERSION="0.14.0"SCALA_MAJOR_VERSION="2.12"

安装大量的 Connector 可能会导致 Jar 包冲突,因此,建议做好如下两项准备工作:

1. 安装新的 Connector 或依赖包时,提前备份一版当前的 lib 库

sudo -u flink cp -r /usr/lib/flink/lib /usr/lib/flink/lib.$(date +'%Y%m%d%H%M').bak

2. 为了解决版本冲突,可以充分 Maven 的依赖解析能力,将需要同时安装的 Connector 的 Maven 依赖整合在一起,去 https://jar-download.com/online-maven-download-tool.php 一次性下载解析好的完整依赖包

3. 安装完毕后,务必重启新的 Yarn Session 方能生效

以下是单独安装各个常用 Connector、组件的方法:

1. Flink SQL Kafka Connector

# install flink kafka connector for flink sql client# only run on master node is enough, owner of flink home dir is 'flink' usersudo -u flink wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/${FLINK_VERSION}/flink-sql-connector-kafka-${FLINK_VERSION}.jar -P /usr/lib/flink/lib/

2. Flink Hudi Connector

# install flink hudi connector for flink sql client# only run on master node is enough, owner of flink home dir is 'flink' usersudo -u flink wget https://repo1.maven.org/maven2/org/apache/hudi/hudi-flink${FLINK_MAJOR_VERSION}-bundle/${HUDI_VERSION}/hudi-flink${FLINK_MAJOR_VERSION}-bundle-${HUDI_VERSION}.jar -P /usr/lib/flink/lib/

3. Flink Hive Connector ( on AWS EMR )

如果 EMR 版本不是 6.15, 请注意替换以下 jar 包文件中的版本号,以所用 EMR 集群上的文件版本为准:

# install flink hive connector for flink sql client# only run on master node is enough, owner of flink home dir is 'flink' user# refer to this doc: https://docs.aws.amazon.com/emr/latest/ReleaseGuide/flink-configure.htmlsudo -u flink cp /usr/lib/hive/lib/antlr-runtime-3.5.2.jar /usr/lib/flink/lib sudo -u flink cp /usr/lib/hive/lib/hive-exec-3.1.3*.jar /usr/lib/flink/lib sudo -u flink cp /usr/lib/hive/lib/libfb303-0.9.3.jar /usr/lib/flink/lib sudo -u flink cp /usr/lib/flink/opt/flink-connector-hive_${SCALA_MAJOR_VERSION}-${FLINK_VERSION}-amzn-1.jar /usr/lib/flink/lib

4. Debezium Confluent Avro 格式 (‘format’ = ‘debezium-avro-confluent’)

前往 https://jar-download.com/online-maven-download-tool.php ,输入如下 Maven 依赖(注意:如有其他同方式获取Jar包的组件,请合并为一份xml配置统一提交,避免出现 Jar 包版本冲突):

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-avro-confluent-registry</artifactId><version>1.18.1</version></dependency>

点击 “Submit” 按钮,将 flink-avro-confluent-registry 及其依赖包下载到本地,然后将得到 jar_files.zip 包上传到集群主节点,并执行以下命令将 jar 包部署到 Flink SQL Client 的 lib 目录下:

# install flink 'debezium-avro-confluent' format for flink sql client# only run on master node is enough, owner of flink home dir is 'flink' user# refer to this doc: https://blog.csdn.net/bluishglc/article/details/135863249 , section 3.2sudo -u flink unzip jar_files.zip -d /usr/lib/flink/lib/

更多详细介绍请参考《Flink 集成 Debezium Confluent Avro ( format=debezium-avro-confluent )》 一文的 3.2 节。

5. Flink JDBC Connector for MySQL

需要同时安装 flink-connector-jdbc 的 Jar 包和 MySQL 的 JDBC 驱动 Jar 包。

# install flink jdbc connector for flink sql client, note: flink-connector-jdbc_2.12-1.14.6.jar is wrong jar!!sudo -u flink wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc/3.1.1-${FLINK_MAJOR_VERSION}/flink-connector-jdbc-3.1.1-${FLINK_MAJOR_VERSION}.jar -P /usr/lib/flink/lib/# install mysql jdbc driver 8.3.0wget https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-j-8.3.0.zip -P /tmp/unzip /tmp/mysql-connector-j-8.3.0.zip -d /tmp/sudo -u flink cp /tmp/mysql-connector-j-8.3.0/mysql-connector-j-8.3.0.jar /usr/lib/flink/libls /usr/lib/flink/lib/mysql-connector-j-8.3.0.jar# install mysql jdbc driver 5.1.49# wget https://cdn.mysql.com/archives/mysql-connector-java-5.1/mysql-connector-java-5.1.49.zip -P /tmp/# unzip /tmp/mysql-connector-java-5.1.49.zip -d /tmp/# sudo -u flink cp /tmp/mysql-connector-java-5.1.49/mysql-connector-java-5.1.49.jar /usr/lib/flink/lib

6. Flink MySQL CDC Connector (2.3.0)

Flink CDC 2.3.0 在官方Repo: https://github.com/ververica/flink-cdc-connectors/tags 上提供了 Uber Jar 供直接下载使用:

# install flink kafka connector for flink sql client# only run on master node is enough, owner of flink home dir is 'flink' usersudo -u flink wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.3.0/flink-sql-connector-mysql-cdc-2.3.0.jar -P /usr/lib/flink/lib/

6. Flink MySQL CDC Connector (2.4+)

注意:目前 Flink CDC 2.4+ 在官方Repo:https://github.com/ververica/flink-cdc-connectors/tags 上尚未提供制作好的 Uber Jar,如果前往 https://jar-download.com/online-maven-download-tool.php 自行制作 Jar 包 + 依赖包,部署后,会报如下错误:

[ERROR] Could not execute SQL statement. Reason:
java.lang.NoSuchMethodError: io.debezium.connector.mysql.MySqlConnection$MySqlConnectionConfiguration.(Lio/debezium/config/Configuration;Ljava/util/Properties;)V

该问题的解释和解决方法参见:https://github.com/ververica/flink-cdc-connectors/issues/2423,鉴于自行重新编译并构建 Uber 包较为繁琐,可先使用 2.3.0 版本,该问题未来会修复。

7. Table Planner 和 Table Planner 加载器

从 Flink 1.15 开始,发行版包含两个 planner: flink-table-planner flink-table-planner-loader。这两个 planner JAR 文件的代码功能相同,但打包方式不同。若使用第一个文件,您必须使用与其相同版本的 Scala;若使用第二个,由于 Scala 已经被打包进该文件里,您不需要考虑 Scala 版本问题。

所以,有时候,我们需要在这两个 Planner 之间进行切换,以下脚本可以完成相互的切换工作:

# install flink-table-planner ( can only choose 1 between flink-table-planner & flink-table-planner-loader )sudo -u flink mv /usr/lib/flink/lib/flink-table-planner-loader-${FLINK_VERSION}-amzn-1.jar /usr/lib/flink/lib/flink-table-planner-loader-${FLINK_VERSION}-amzn-1.jar.baksudo -u flink cp /usr/lib/flink/opt/flink-table-planner_${SCALA_MAJOR_VERSION}-${FLINK_VERSION}-amzn-1.jar /usr/lib/flink/lib# revert to flink-table-planner-loader ( can only choose 1 between flink-table-planner & flink-table-planner-loader ) sudo -u flink rm -f /usr/lib/flink/opt/flink-table-planner_${SCALA_MAJOR_VERSION}-${FLINK_VERSION}-amzn-1.jarsudo -u flink mv /usr/lib/flink/lib/flink-table-planner-loader-${FLINK_VERSION}-amzn-1.jar.bak /usr/lib/flink/lib/flink-table-planner-loader-${FLINK_VERSION}-amzn-1.jar

常见问题

1. [ERROR] Could not execute SQL statement. Reason:
java.lang.NoSuchMethodError: io.debezium.connector.mysql.MySqlConnection$MySqlConnectionConfiguration.(Lio/debezium/config/Configuration;Ljava/util/Properties;)V

该问题的解释和解决方法参见:https://github.com/ververica/flink-cdc-connectors/issues/2423,鉴于自行重新编译并构建 Uber 包较为繁琐,可先使用 2.3.0 版本,该问题未来会修复。