文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Apache Flink 扫雷系列 - PyFlink如何解决多JAR包依赖问题

2024-12-24 18:13

关注

本篇是《Apache Flink 扫雷系列》的第一篇,所以简单介绍一下这一系列的特点,本系列所定义的”雷”是指由于目前Apache Flink目前的设计问题导致的用户非便利性问题的临时解决办法。那么为什么明知道有设计问题还不进行设计重构,避免这些”雷”的存在呢?其实社区的发展和我们各个公司内部产品发展一样,都有一些客观因素导致实际存在的问题无法及时得到解决,比如,社区的Release或者内部产品发布的的周期问题,在没有新的Release之前的一些对用户非友好的问题就需要有一些“非正规”的解决方式,或者说是临时解决方案,这种方案的特点就是,能解决问题,但不是通用性解决手段,只能民间流传,不能官方宣扬。所以《Apache Flink 扫雷系列》就是为大家提供能够解决大家现实问题,但是可能不是最佳实践,大家在这系列中可以有更大的反哺社区的机会:)

开篇说”雷”

本篇的”雷”是目前针对Apache Flink 1.10集以前版本中,在利用CLI提交作业时候只能提交一个JAR的功能问题解决,也就是针对命令参数-j,--jarfile Flink program JAR file. 的问题。目前 -j只允许用户提供一个JAR包,这在很多场景是不太合理的,不用说用户自己的JAR包,就单说Flink用户使用的Connector的JAR,在一个作业里面就可以能使用多个不同的Connector类型,比如在《PyFlink 场景案例 - PyFlink实现CDN日志实时分析》为例,就用了Kafka,MySql,CSV等多种功能JAR包的依赖。这个普遍的问题就是要用手工方式将这三个JARs合并成一个,然后提交作业时候用-j选项上传到集群。

扫雷难度

面对合并多个JAR包,也许Java用户还好(虽然不便利,但应该都会操作),但对于Python用户,在没有涉及过Java开发的情况下,可能要花费一些时间来完成JARs的合并,甚至有可能有种无从下手的感觉。所以本篇主要针对的是不了解Java的Flink Python用户。

案例选取

为了大家能够实际的体验实际效果,我们选取一个具体的案例来说明如果进行多JARs的合并。我们就选取我在2020年3月17日直播中所说的《PyFlink 场景案例 - PyFlink实现CDN日志实时分析》来进行说明。

案例回顾

PyFlink 场景案例 - PyFlink实现CDN日志实时分析》核心是针对灌入Kafka的CDN日志数据经过PyFlink进行按地区的下载量,下载速度的统计,最终将统计数据写入到MySql中。同时放入到Kafka的数据格式是CSV('format.type' = 'csv')。所以我们依赖的JARs如下:

我们可以用如下命令下载:

  1. $ curl -O https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka_2.11/1.10.0/flink-sql-connector-kafka_2.11-1.10.0.jar 
  2. $ curl -O https://repo1.maven.org/maven2/org/apache/flink/flink-jdbc_2.11/1.10.0/flink-jdbc_2.11-1.10.0.jar 
  3. $ curl -O https://repo1.maven.org/maven2/org/apache/flink/flink-csv/1.10.0/flink-csv-1.10.0-sql-jar.jar 
  4. $ curl -O https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.19/mysql-connector-java-8.0.19.jar 

我们将如上4个JARs下载到某个目录,我这里下载到本机的temp目录:

“雷”存在的场景说明

为啥在博客《PyFlink 场景案例 - PyFlink实现CDN日志实时分析》并没有提到要合并JARs的问题? 是的,这个“雷”的存在是有一定的条件的:

作业提交的集群环境没有预先安装你所有需要的JARs(大部分情况都是不会安装的)

上面条件是必须成立,才会存在扫雷的问题。那么在博客中我在集群环境预安装了说需要的JARs,也就是博客中提到的下载JARs到集群lib目录

  1. PYFLINK_LIB=python -c "import pyflink;import os;print(os.path.dirname(os.path.abspath(pyflink.__file__))+'/lib')") 

的操作。

合并JARs的注意点

合并JARs的一个很重要的点是涉及到了JAR包的Service Provider机制,详细规范详见。这是让Python人员是很难注意到的合并重点。JAR包的Service Provider机制会允许在JAR包的META-INF/services目录下保存Service Provider的配置文件。简单说就是他为开发者提供了一种扩展机制,在开发阶段只是定义接口,然后在包含实现的JAR包进行实现配置,就可以调用到实际接口的实现类。关于JAR包META-INF目录结构简单说明如下:

注意:provider配置文件必须是以UTF-8编码。

合并操作

1. 解压JARs

  1. $ mkdir jobjar csv jdbc kafka mysql 

其中jobjar存放最终我们打包的JAR内容, csv jdbc kafka mysql存放对应的JAR所解压的内容。具体命令如下:

  1. $ unzip flink-csv-1.10.0-sql-jar.jar -d csv/ 
  2. $ unzip flink-sql-connector-kafka_2.11-1.10.0.jar -d kafka/ 
  3. $ unzip flink-jdbc_2.11-1.10.0.jar -d jdbc/ 
  4. $ unzip mysql-connector-java-8.0.19.jar -d mysql 

解压之后我们会在刚才的目录得到如下文件内容:

我们核心要处理的是class文件夹和 META-INF/services文件夹,如图csv和kafka的JAR解压之后的内容。其中,Class文件夹可以直接拷贝。但是services要进行同名的合并,比如上用于Flink的Connector的服务发现配置org.apache.flink.table.factories.TableFactory是需要将文件内容进行合并的。

2. 合并JARs

首先我们创建META-INF和META-INF/services目录,目录结构如下:

  1. jincheng:jobjar jincheng.sunjc$ tree -L 2 
  2. └── META-INF 
  3.     └── services 
  4. 2 directories, 0 files 

(1) class文件合并

将csv jdbc kafka mysql的class直接copy到jobjar目录,如下:

  1. $ cp -rf ../csv/org . 
  2. $ cp -rf ../jdbc/org . 
  3. $ cp -rf ../kafka/org . 
  4. $ cp -rf ../mysql/com . 
  5. $ tree -L 2 
  6.   . 
  7.   ├── META-INF 
  8.   │   └── services 
  9.   ├── com 
  10.   │   └── mysql 
  11.   └── org 
  12.       └── apache 

详细的目录结构如下:

(2) services合并

Service Provider是JAR的一个标准,不仅仅Flink的Connector使用了Service Provider机制,同时Kafka使用了配置的服务发现。所以我们要将所有的services里面的内容按文件名进行合并。以csv和kafka为例:

在CSV里面的META-INF/services里面只有一个和Flink的connector相关的配置,内容如下:

在Kafka里面的META-INF/services里面有Flink的connector相关的配置和Kafka内部使用的配置,内容如下:

所以我们需要将Kafka相关的直接copy到jobjar/META-INF/services/目录,然后将csv和Kafka关于org.apache.flink.table.factories.TableFactory的配置进行内容合并。合并的内容如下:

  1. # Licensed to the Apache Software Foundation (ASF) under 
  2. ... 
  3. ... 
  4. # limitations under the License. 
  5.  
  6. org.apache.flink.formats.csv.CsvRowFormatFactory 
  7. org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactory 

我们最终将4个JARs的services配置进行合并之后的最终代码如下:

  1. # Licensed to the Apache Software Foundation (ASF) under 
  2. ... 
  3. ... 
  4. # limitations under the License. 
  5.  
  6. org.apache.flink.formats.csv.CsvRowFormatFactory 
  7. org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactory 
  8. org.apache.flink.api.java.io.jdbc.JDBCTableSourceSinkFactory 

大家可以尝试使用的命令如下:

  1. $ cat ../csv/META-INF/services/org.apache.flink.table.factories.TableFactory | grep ^[^#] >> META-INF/services/org.apache.flink.table.factories.TableFactory 
  2. $ cat ../kafka/META-INF/services/org.apache.flink.table.factories.TableFactory | grep ^[^#] >> META-INF/services/org.apache.flink.table.factories.TableFactory 
  3. $ cat ../kafka/META-INF/services/org.apache.flink.kafka.shaded.org.apache.kafka.common.config.provider.ConfigProvider | grep ^[^#] >> META-INF/services/org.apache.flink.kafka.shaded.org.apache.kafka.common.config.provider.ConfigProvider 
  4. $ cat ../jdbc/META-INF/services/org.apache.flink.table.factories.TableFactory | grep ^[^#] >> META-INF/services/org.apache.flink.table.factories.TableFactory 

3. 创建JAR

这一步骤没有特别强调的内容,直接用用zip或者jar命令进行打包就好了。

  1. $ jincheng:jobjar jincheng.sunjc$ jar -cf myjob.jar META-INF com org 

我最终产生的JAR可以在这里下载,用于对比你自己打包的是否和我的一样:)

OK,到这里我们就完成了多JARs的合并工作。我们可以尝试应用CLI进行提交命令了。

CLI提交作业

当没有添加-j选项时候,提交作业如下:

  1. $PYFLINK_LIB/../bin/flink run -m localhost:4000 -py cdn_demo.py 

报错如下:

提供正确的-j参数,将我们打包的JAR提交到集群的情况,如下:

  1. $PYFLINK_LIB/../bin/flink run -j ~/temp/jobjar/myjob.jar -m localhost:4000 -py cdn_demo.py 

同时Web控制台可以查看提交的作业:

小结

本篇核心介绍了PyFlink的用户如何解决多JARs依赖作业提交问题,也许这不是最Nice的解决方法,但至少是你解决多JARs依赖作业提交的方法之一,祝你 “扫雷” 顺利,也期望如果你有更好的解决办法,留言或者邮件与我分享哦:)!

【本文为51CTO专栏作者“金竹”原创稿件,转载请联系原作者】

戳这里,看该作者更多好文 

 

来源:51CTO专栏内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯