博主历时三年精心创作的《大数据平台架构与原型实现:数据中台建设实战》一书现已由知名IT图书品牌电子工业出版社博文视点出版发行,点击《重磅推荐:建大数据平台太难了!给我发个工程原型吧!》了解图书详情,京东购书链接:https://item.jd.com/12677623.html,扫描左侧二维码进入京东手机购书页面。 |
文章目录
- 1. Flink SQL Kafka Connector
- 2. Flink Hudi Connector
- 3. Flink Hive Connector ( on AWS EMR )
- 4. Debezium Confluent Avro 格式 ('format' = 'debezium-avro-confluent')
- 5. Flink JDBC Connector for MySQL
- 6. Flink MySQL CDC Connector (2.3.0)
- 7. Flink MySQL CDC Connector (2.4+)
- 8. Flink Changlog 格式 ('format' = 'changelog-json')
- 9. Table Planner 和 Table Planner 加载器
- 常见问题
一般来说,在 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 一次性下载解析好的完整依赖包
另外,要特别提醒的是: 安装完毕后,务必重启新的 Yarn Session 方能生效。以下是单独安装各个常用 Connector、组件的方法:
1. Flink SQL Kafka Connector
```bash # install flink kafka connector for flink sql client # only run on master node is enough, owner of flink home dir is 'flink' user sudo -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' user
sudo -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.html
sudo -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.2
sudo -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.0
wget 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/lib
ls /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' user
sudo -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/
7. 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 版本,该问题未来会修复。
8. Flink Changlog 格式 (‘format’ = ‘changelog-json’)
sudo -u flink wget https://repo1.maven.org/maven2/com/ververica/flink-format-changelog-json/2.1.1/flink-format-changelog-json-2.1.1.jar -P /usr/lib/flink/lib/
安装后,重启 sql client 可能会报:java.lang.ClassNotFoundException: org.apache.flink.formats.json.JsonOptions 错误,应该是却少依赖包,由于后来并没有真正使用到该格式,所以此文暂时搁置。
9. 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.bak
sudo -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.jar
sudo -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 版本,该问题未来会修复。