Datastream 开发打包问题

简介: Datastream作业开发时往往会遇到一些jar包冲突等问题,本文主要讲解作业开发时需要引入哪些依赖以及哪些需要被打包进作业的jar中,从而避免不必要的依赖被打入了作业jar中以及可能产生的依赖冲突。

Datastream作业开发时往往会遇到一些jar包冲突等问题,本文主要讲解作业开发时需要引入哪些依赖以及哪些需要被打包进作业的jar中,从而避免不必要的依赖被打入了作业jar中以及可能产生的依赖冲突。

一个Datastream作业主要涉及下述依赖:

Flink的核心依赖以及应用程序自身的依赖

每一个Flink应用程序都依赖于一系列相关的库,其中至少应该包括Flink的API. 许多应用程序还依赖于连接器相关的库(如 Kafka, Cassandra等).在运行Flink应用程序时,无论是在运行在分布式的环境下还是在本地IDE进行测试,Flink的运行时相关依赖都是必须的。

与大多数运行用户自定义应用程序的系统一样,Flink 中有两大类依赖项:

  • Flink核心依赖:Flink 本身由一组运行系统所必需的类和依赖项组成,例如协调器、网络、检查点、容错、API、算子(例如窗口)、资源管理等。 所有这些类和依赖项的集合构成了 Flink 运行时的核心,在 Flink 应用程序启动时必须存在。这些核心类和依赖项都被打包在 flink-dist jar 中。 它们是 Flink 的 lib 文件夹的一部分,也是Flink基础容器镜像的一部分。这些依赖之于Flink就像Java 运行所需的包含 String 和 List 等类的核心库(rt.jar、charsets.jar 等)之于Java。Flink的核心依赖不包含任何连接器或扩展库(CEP、SQL、ML等),这使得Flink的核心依赖尽可能小,以避免默认情况下类路径中有过多的依赖项,同时减少依赖冲突。
  • 用户应用程序依赖项:指特定用户应用程序所需的所有连接器、Format或扩展库。用户应用程序通常被打包成一个 jar文件,其中包含应用程序代码以及所需的连接器和库依赖项。用户应用程序依赖项不应包括 Flink DataStream API 和运行时依赖项,因为这些已经被包含在了Flink 的核心依赖中。

依赖配置步骤

1.添加基础依赖

每一个Flink应用程序的开发至少需要添加对相关API的基础依赖。

手动配置项目时,需要添加对Java/Scala API的依赖(这里以Maven为例,在其他构建工具(Gradle,SBT等)中可以使用同样的依赖)。

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>1.12.3</version>
  <scope>provided</scope>
</dependency>

重要提示:请注意,所有这些依赖项都将其范围设置为"provided"。这意味着需要对它们进行编译,但不应将它们打包到项目生成的应用程序jar文件中——这些依赖项是Flink核心依赖项,在实际运行时已经被加载。

强烈建议将依赖项设置成"provided"的范围,如果未将它们设置为"provided",最好的情况下会导致生成的jar变得臃肿,因为它还包含所有Flink核心依赖项。而最怀的情况下,添加到应用程序jar文件中的Flink核心依赖项与您自己的一些依赖项会发生版本冲突(通常通过Flink的反向类加载机制来避免)。

关于IntelliJ的注意事项:为了使应用程序在IntelliJ IDEA中运行,有必要在运行配置中勾选"Include dependencies with "Provided" scope"选项框。如果没有该选项(可能是由于使用较旧的IntelliJ IDEA版本),那么一个简单的解决方法是创建一个调用应用程序 main() 方法的测试用例。

2.添加连接器和库的依赖

大多数应用程序的运行需要特定的连接器或库,例如Kafka、Cassandra等连接器。这些连接器不是Flink核心依赖项的一部分,必须作为额外依赖项添加到应用程序中。

下述代码是添加Kafka连接器依赖项的示例(Maven语法):

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>1.12.3</version>
</dependency>

我们建议将应用程序代码和它所有的依赖以jar-with-dependencies 的形式打包到一个application jar中。这个应用程序jar包可以被提交到已经存在的Flink集群上去,或者被加入到Flink应用程序的容器镜像中去。

从Maven作业模版(见下文Maven作业模版部分)创建的项目,通过mvn clean package命令会自动把依赖打到应用程序的jar包中去。对于没有使用模版进行配置的情况,建议使用Maven Shade Plugin (配置如附录所示) 来构建包含依赖的jar包。

重要提示:对于Maven(和其他构建工具)来说,要将依赖项正确打包到应用程序jar中,这些应用程序依赖项的scope必须指定为"compile"(与核心依赖项不同,核心依赖项的scope必须指定为"provided")。

注意事项

Scala版本

Scala的不同版本(2.11,2.12等)相互之间是不兼容的。因此,Scala 2.11对应的Flink版本不能用于使用Scala 2.12的应用程序。

所有依赖(或传递依赖)于Scala的Flink依赖项都以构建它们的Scala版本作为后缀,例如flink-streaming-scala_2.11。

只使用Java进行开发时可以选择任何Scala版本,使用Scala开发时需要选择与其应用程序的Scala版本匹配的Flink依赖版本。

:2.12.8之后的Scala版本与之前的2.12.x版本不兼容,因此Flink项目无法将其2.12.x版本升级到2.12.8之后的版本。用户可以在本地自己编译对应Scala版本的Flink。为了使其能够正常工作,需要添加-Djapicmp.skip以在构建时跳过二进制兼容性检查。

Hadoop依赖

一般的规则: 永远不要将Hadoop相关依赖直接添加到应用程序中. (唯一的例外是将现有的Hadoop输入/输出Format与Flink的Hadoop兼容包一起使用时)

如果希望将Flink与Hadoop结合使用,则需要包含Hadoop依赖的Flink启动项,而不是将Hadoop添加为应用程序依赖项。Flink将使用HADOOP_CLASSPATH环境变量指定的Hadoop依赖项,可通过以下方式进行设置:

export HADOOP_CLASSPATH**=**hadoop classpath``

这种设计有两个主要原因:

  • 一些与Hadoop的交互可能发生在Flink的核心模块中,并且在用户应用程序启动之前,例如为检查点设置HDFS、通过Hadoop的Kerberos令牌进行身份验证,或者在YARN上进行部署等。
  • Flink的反向类加载机制从核心依赖项中隐藏了许多可传递的依赖项。这不仅适用于Flink自己的核心依赖项,而且适用于Hadoop的依赖项。这样,应用程序就可以使用相同依赖项的不同版本,而不会发生依赖项冲突(相信我们,这是一件大事,因为Hadoop依赖树非常庞大。)

如果在IDE内部的测试或开发过程中需要Hadoop依赖项(例如HDFS访问),请将这些依赖项的scope配置为

test 或则 provided

Transform table connector/format resources #

Flink使用Java的Service Provider Interfaces (SPI) 机制通过特定标识符加载table的connector/format工厂。由于每个table的connector/format的名为org.apache.flink.table.factories.Factory的SPI资源文件位于同一目录:META-INF/services下,因此在构建使用多个table connector/format的项目的uber jar时,这些资源文件将相互覆盖,这将导致Flink无法正确加载工厂类。

在这种情况下,推荐的方法是通过maven shade插件的ServicesResourceTransformer转换META-INF/services目录下的这些资源文件。给定示例的pom.xml文件内容如下,其中包含连接器flink-sql-connector-hive-3.1.2和flink-parquet format。

    <modelVersion>4.0.0</modelVersion>
    <groupId>org.example</groupId>
    <artifactId>myProject</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <!--  other project dependencies  ...-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-sql-connector-hive-3.1.2__2.11</artifactId>
            <version>1.13.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-parquet__2.11<</artifactId>
            <version>1.13.0</version>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <executions>
                    <execution>
                        <id>shade</id>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers combine.children="append">
                                <!-- The service transformer is needed to merge META-INF/services files -->
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                <!-- ... -->
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

在配置了ServicesResourceTransformer之后, 项目构建uber-jar时,META-INF/services目录下的这些资源文件会被整合在一起而不是相互覆盖。

Maven作业模版

强烈建议使用该方式进行配置,可以减少很多重复的配置工作。

前置要求

唯一的环境要求是安装了Maven 3.0.4(或更高版本)和Java 8.x。

创建项目

使用以下两种方式中的一种创建项目:

  • 使用Maven archetypes
$ mvn archetype:generate                               \
  -DarchetypeGroupId=org.apache.flink              \
  -DarchetypeArtifactId=flink-quickstart-java      \
  -DarchetypeVersion=1.12.3

这允许您命名新创建的项目。它将以交互方式要求您输入groupId、artifactId和包名。

  • 运行quickstart脚本
$ curl https://flink.apache.org/q/quickstart.sh | bash -s 1.12.3

我们建议您将此项目导入IDE以开发和测试它。IntelliJ IDEA原生支持Maven项目。如果使用Eclipse,可以使用m2e插件导入Maven项目。默认情况下,某些Eclipse捆绑包包含该插件,否则需要您手动安装。

请注意:默认的Java JVM heap size对于Flink来说可能太小了。你必须手动增加它。在Eclipse中,选择RunConfigurations->Arguments并写入VM Arguments框:-Xmx800m。在IntelliJ IDEA中,更改JVM选项的推荐方法是使用Help | Edit Custom VM Options选项菜单。

构建项目

如果要生成/打包项目,请转到项目目录并运行"mvn clean package"命令。执行后将会得到一个JAR文件:target/-.jar,其中包含您的应用程序,以及作为依赖项添加到应用程序的连接器和库。

注意:如果使用与StreamingJob不同的类作为应用程序的主类/入口点,我们建议您相应地更改pom.xml文件中的mainClass设置。这样,Flink就可以直接从JAR文件运行应用程序,而无需另外指定主类。

附录: 构建带依赖的jar包的模版

要构建包含连接器和库所需的所有依赖项的应用程序JAR,可以使用以下shade插件定义:

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.1.1</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <artifactSet>
                            <excludes>
                                <exclude>com.google.code.findbugs:jsr305</exclude>
                                <exclude>org.slf4j:*</exclude>
                                <exclude>log4j:*</exclude>
                            </excludes>
                        </artifactSet>
                        <filters>
                            <filter>
                                <!-- Do not copy the signatures in the META-INF folder.
                                Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                <artifact>*:*</artifact>
                                <excludes>
                                    <exclude>META-INF/*.SF</exclude>
                                    <exclude>META-INF/*.DSA</exclude>
                                    <exclude>META-INF/*.RSA</exclude>
                                </excludes>
                            </filter>
                        </filters>
                        <transformers>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                <mainClass>my.programs.main.clazz</mainClass>
                            </transformer>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

更多 Flink 相关技术问题,可扫码加入社区钉钉交流群
第一时间获取最新技术文章和社区动态,请关注公众号~

image.png

活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算Flink版现开启活动:
99 元试用 实时计算Flink版(包年包月、10CU)即有机会获得 Flink 独家定制卫衣;另包 3 个月及以上还有 85 折优惠!
了解活动详情:https://www.aliyun.com/product/bigdata/sc

image.png

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
消息中间件 SQL Java
Flink自定义Connector
Flink自定义Connector
950 0
|
SQL 缓存 Java
flink cdc 同步问题之如何同步多张库表
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
XML SQL Java
Maven的三种打包方式(jar、shade、assembly)
Maven的三种打包方式(jar、shade、assembly)
6129 0
|
消息中间件 缓存 监控
Flink背压原理以及解决优化
Flink背压原理以及解决优化
1132 0
|
SQL 存储 数据库
Flink + Paimon 数据 CDC 入湖最佳实践
Flink + Paimon 数据 CDC 入湖最佳实践
2924 59
|
8月前
|
SQL 关系型数据库 MySQL
Flink CDC 3.4 发布, 优化高频 DDL 处理,支持 Batch 模式,新增 Iceberg 支持
Apache Flink CDC 3.4.0 版本正式发布!经过4个月的开发,此版本强化了对高频表结构变更的支持,新增 batch 执行模式和 Apache Iceberg Sink 连接器,可将数据库数据全增量实时写入 Iceberg 数据湖。51位贡献者完成了259次代码提交,优化了 MySQL、MongoDB 等连接器,并修复多个缺陷。未来 3.5 版本将聚焦脏数据处理、数据限流等能力及 AI 生态对接。欢迎下载体验并提出反馈!
1430 1
Flink CDC 3.4 发布, 优化高频 DDL 处理,支持 Batch 模式,新增 Iceberg 支持
|
10月前
|
SQL 存储 消息中间件
vivo基于Paimon的湖仓一体落地实践
本文整理自vivo互联网大数据专家徐昱在Flink Forward Asia 2024的分享,基于实际案例探讨了构建现代化数据湖仓的关键决策和技术实践。内容涵盖组件选型、架构设计、离线加速、流批链路统一、消息组件替代、样本拼接、查询提速、元数据监控、数据迁移及未来展望等方面。通过这些探索,展示了如何优化性能、降低成本并提升数据处理效率,为相关领域提供了宝贵的经验和参考。
1143 3
vivo基于Paimon的湖仓一体落地实践
|
11月前
|
消息中间件 关系型数据库 MySQL
基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
本教程展示如何使用Flink CDC YAML快速构建从MySQL到Kafka的流式数据集成作业,涵盖整库同步和表结构变更同步。无需编写Java/Scala代码或安装IDE,所有操作在Flink CDC CLI中完成。首先准备Flink Standalone集群和Docker环境(包括MySQL、Kafka和Zookeeper),然后通过配置YAML文件提交任务,实现数据同步。教程还介绍了路由变更、写入多个分区、输出格式设置及上游表名到下游Topic的映射等功能,并提供详细的命令和示例。最后,包含环境清理步骤以确保资源释放。
837 2
基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错之如何解决报错:Could not acquire the minimum required resources.
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
Java 关系型数据库 数据库连接
实时计算 Flink版操作报错之遇到错误org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'jdbc',该如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。