Re: Flink write ADLS show error: No FileSystem for scheme "file"
Hi, Gabor,
I'm curious about why this happened in Azure file and not in other file
format(I tried use s3 and it works OK)
Gabor Somogyi 于2024年7月2日周二 16:59写道：
> I see, thanks for sharing.
>
> The change what you've made makes sense. Let me explain the details.
> Each and every plugin has it's own class loader. The reason behind that is
> to avoid dependency collision with Flink's main class loader.
>
> I think if the mentioned change works when it's added as normal lib and
> not as a plugin then the code can be merged to main as-is.
>
> G
>
>
> On Thu, Jun 27, 2024 at 5:30 AM Xiao Xu wrote:
>
>> Hi, Gabar,
>>
>> Thanks to reply, I make sure that not conflict in maven, all the hadoop
>> dependency is in provided scope,
>> and checked my result jar it not contains
>> (src/main/resources/META-INF/services).
>>
>> This is my pom:
>>
>> http://maven.apache.org/POM/4.0.0;
>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance;
>> xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
>> http://maven.apache.org/xsd/maven-4.0.0.xsd;>
>> 4.0.0
>>
>> com.test.flink
>> flink-sync
>> 1.0-SNAPSHOT
>> jar
>>
>> Flink Quickstart Job
>>
>>
>> 1.8
>> 1.8
>> 1.18.1
>> 1.8
>> 2.12
>> 3.2.0
>>3.3.4
>>2.16.0
>>3.2.0
>>
>>
>>
>>
>> org.apache.flink
>> flink-java
>> ${flink.version}
>> provided
>>
>>
>>
>> org.apache.flink
>> flink-streaming-java
>> ${flink.version}
>> provided
>>
>>
>>
>> org.apache.flink
>> flink-clients
>> ${flink.version}
>> provided
>>
>>
>>
>> org.apache.flink
>> flink-connector-files
>> ${flink.version}
>>
>>
>> org.apache.flink
>> flink-connector-kafka
>> 3.1.0-1.18
>>
>>
>>
>> org.apache.logging.log4j
>> log4j-slf4j-impl
>> ${log4j.version}
>> runtime
>>
>>
>> slf4j-api
>> org.slf4j
>>
>>
>>
>>
>> org.apache.logging.log4j
>> log4j-api
>> ${log4j.version}
>> runtime
>>
>>
>> org.apache.logging.log4j
>> log4j-core
>> ${log4j.version}
>> runtime
>>
>>
>>
>> org.apache.flink
>> flink-azure-fs-hadoop
>> ${flink.version}
>> provided
>>
>>
>>
>>
>>
>> org.apache.maven.plugins
>> maven-assembly-plugin
>> 3.0.0
>>
>> false
>>
>>jar-with-dependencies
>>
>>
>>
>>
>>make-assembly
>>package
>>
>> single
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> And like my reply in stackoverflow, I found the hadoop-common file :
>> https://github.com/apache/hadoop/blob/release-3.3.4-RC1/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java#L3374
>> do not load any filesystem, dig in ServiceLoader.load(FileSystem.class)
>> source code, it looks like have different class loader make it not load
>> any filesystem.
>> I changed the ServiceLoader.load(FileSystem.class) to
>> ServiceLoader.load(FileSystem.class,
>> FileSystem.class.getClassLoader()) and replace the flink-fs-azure-hadoop
>> plugin, it works now,
>> So I'm not sure why it works
>>
>> Gabor Somogyi 于2024年6月26日周三 16:52写道：
>>
>>> Hi Xiao,
>>>
>>> I'm not quite convinced that the azure plugin ruined your workload, I
>>> would take a lo
Re: Flink write ADLS show error: No FileSystem for scheme "file"
Hi, Gabar,
Thanks to reply, I make sure that not conflict in maven, all the hadoop
dependency is in provided scope,
and checked my result jar it not contains
(src/main/resources/META-INF/services).
This is my pom:
http://maven.apache.org/POM/4.0.0;
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance;
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd;>
4.0.0
com.test.flink
flink-sync
1.0-SNAPSHOT
jar
Flink Quickstart Job
1.8
1.8
1.18.1
1.8
2.12
3.2.0
3.3.4
2.16.0
3.2.0
org.apache.flink
flink-java
${flink.version}
provided
org.apache.flink
flink-streaming-java
${flink.version}
provided
org.apache.flink
flink-clients
${flink.version}
provided
org.apache.flink
flink-connector-files
${flink.version}
org.apache.flink
flink-connector-kafka
3.1.0-1.18
org.apache.logging.log4j
log4j-slf4j-impl
${log4j.version}
runtime
slf4j-api
org.slf4j
org.apache.logging.log4j
log4j-api
${log4j.version}
runtime
org.apache.logging.log4j
log4j-core
${log4j.version}
runtime
org.apache.flink
flink-azure-fs-hadoop
${flink.version}
provided
org.apache.maven.plugins
maven-assembly-plugin
3.0.0
false
jar-with-dependencies
make-assembly
package
single
And like my reply in stackoverflow, I found the hadoop-common file :
https://github.com/apache/hadoop/blob/release-3.3.4-RC1/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java#L3374
do not load any filesystem, dig in ServiceLoader.load(FileSystem.class)
source code, it looks like have different class loader make it not load
any filesystem.
I changed the ServiceLoader.load(FileSystem.class) to
ServiceLoader.load(FileSystem.class,
FileSystem.class.getClassLoader()) and replace the flink-fs-azure-hadoop
plugin, it works now,
So I'm not sure why it works
Gabor Somogyi 于2024年6月26日周三 16:52写道：
> Hi Xiao,
>
> I'm not quite convinced that the azure plugin ruined your workload, I
> would take a look at the dependency graph you've in the pom.
> Adding multiple deps can conflict in terms of class loader services
> (src/main/resources/META-INF/services).
>
> As an example you've 2 such dependencies where
> org.apache.flink.core.fs.FileSystemFactory is in the jar.
> Hadoop core contains "flie" and the other one something different. Let's
> say you don't use service merge plugin in your
> maven project. Such case Hadoop core `file` entry will be just overwritten
> by the second one.
>
> Solution: Either avoid deps with conflicting services or add
> ServicesResourceTransformer
> to your maven project.
>
> G
>
>
> On Wed, Jun 26, 2024 at 10:16 AM Xiao Xu wrote:
>
>> Hi, all
>>
>> I try to use Flink to write Azure Blob Storage which called ADLS, I put
>> the flink-azure-fs-hadoop jar in plugins directory and when I start my
>> write job it shows:
>>
>> Caused by: org.apache.hadoop.fs.UnsupportedFileSystemException: No
>> FileSystem for scheme "file"
>> at
>> org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3443)
>> ~[hadoop-common-3.3.4.5.1.5.3.jar:?]
>> at
>> org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3466)
>> ~[hadoop-common-3.3.4.5.1.5.3.jar:?]
>> at
>> org.apache.hadoop.fs.FileSystem.access$300(FileSystem.java:174)
>> ~[hadoop-common-3.3.4.5.1.5.3.jar:?]
>> at
>> org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3574)
>> ~[hadoop-common-3.3.4.5.1.5.3.jar:?]
>> at
>> org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3521)
>> ~[hadoop-common-3.3.4.5.1.5.3.jar:?]
>> at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:540)
>> ~[hadoop-common-3.3.4.5.1.5.3.jar:?]
>>
Flink write ADLS show error: No FileSystem for scheme "file"
Hi, all I try to use Flink to write Azure Blob Storage which called ADLS, I put the flink-azure-fs-hadoop jar in plugins directory and when I start my write job it shows: Caused by: org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "file" at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3443) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3466) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem.access$300(FileSystem.java:174) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3574) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3521) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:540) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem.getLocal(FileSystem.java:496) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.confChanged(LocalDirAllocator.java:316) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.getLocalPathForWrite(LocalDirAllocator.java:393) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.LocalDirAllocator.getLocalPathForWrite(LocalDirAllocator.java:165) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.LocalDirAllocator.getLocalPathForWrite(LocalDirAllocator.java:146) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.store.DataBlocks$DiskBlockFactory.createTmpFileForWrite(DataBlocks.java:980) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.store.DataBlocks$DiskBlockFactory.create(DataBlocks.java:960) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream.createBlockIfNeeded(AbfsOutputStream.java:262) ~[hadoop-azure-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream.(AbfsOutputStream.java:173) ~[hadoop-azure-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore.createFile(AzureBlobFileSystemStore.java:580) ~[hadoop-azure-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem.create(AzureBlobFileSystem.java:301) ~[hadoop-azure-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1195) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1175) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1064) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1052) ~[hadoop-common-3.3.4.5.1.5.3.jar:?] I search the issue looks like this: https://stackoverflow.com/questions/77238642/apache-flink-azure-abfs-file-sink-error-streaming-unsupportedfilesystemexcep my env: Flink: 1.18.1 JDK: 1.8 Does anyone else have the same problem？
Re: kafka table connector保留多久的数据
flink 是不会保留数据的, 数据都是落盘在 kafka 里, flink 根据 offset 去读 kafka 里的数据, 可以设置 kafka 里留存的时间 marble.zh...@coinflex.com.INVALID 于2020年10月14日周三 下午4:37写道： > 你好， 用kafka table > > connector接过来的数据，在flink这边会保留多久，在参数列表里没有看到有这个设置，如果保留太久，内存会撑暴，比如我只想保留半个小时，之前的数据可以清除。 > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/
Re: flink sink kafka过慢问题
建议先确认下瓶颈是不是 kafka sink, 一般来说 kafka 网卡打满都不会到瓶颈的, 猜测有可能其他逻辑导致的瓶颈 hailongwang <18868816...@163.com> 于2020年10月13日周二 下午10:22写道： > > > Hi xyq， > 1. 可以确认下下游 kakfa 6个分区写入数据量都是均匀的吗，看下 Partitioner 有没有设置好。 > 2. 还有 11000 条的数据量大小有多少呢，有没有存在 flink 集群 与 kafka 集群 > 跨机房的限制。（在我们内部多个机房，其中延迟比较大的机房的速率只有 3M/s 单并发） > 3. 可以确认下，Kafka sink 有没有一些耗时的序列化操作的 > 个人认为，这种问题我们最好分而治之，可以测试下 正常情况下两集群间的传输速率（比如scp），然后再看看 kafka 集群，然后 flink 任务等。 > 希望可以帮助到你 > > > Best， > Hailong Wang > > > > > 在 2020-10-13 18:38:37，"xyq" 写道： > >hi，all > > > > > >我在使用flink处理数据写入kafka的过程中，写入kafka的速度过慢，导致数据高峰期数据有堆积， > >在数据达到每秒11000条的时候就开始堆积，处理逻辑很简单，就有坐标点经纬度转换成省市区，其他逻辑不复杂， > >sink端kafka 6个分区，source端kafka6个分区，每天在晚高峰的时候会堆积30分钟，有没有办法大幅度提高写入能力， > >难道只能加大kafka的分区吗？ > > > > > >flink submit 参数如下 > >p=6（与kafka分区对应） > >m=yarn-cluster > >yjm=1024 > >ytm=4096 > >ys=6 > >checkpoint_timeout=3 > >max_concurrent_checkpoint=1 > >checkpoint_between_time=1000 > >checkpoint_interval_ms=6 > >restart_between_time=1000 > >restart_num=4 > >sink_buffer_memory=134217728（也尝试了很多） > >sink_batch_size=65536（也尝试了很多） > >sink_kafka_linger_ms=10（尝试过1 20 50 100 200 500） > >sink_max_request_size=1024 > >sink_kafka_acks=1 > > > > > >这个问题困扰我很久了，请大家帮忙看看 ，非常感谢！ >
Re: 如何获取flink webUI上面的DAG图
plan visualizer 应该是 stream graph, 不是一个图吧 Paul Lam 于2020年10月13日周二 下午9:23写道： > Hi, > > 可以利用 Flink 的 plan visualizer，见[1] > > 1. > > https://ci.apache.org/projects/flink/flink-docs-stable/dev/execution_plans.html > > Best, > Paul Lam > > hailongwang <18868816...@163.com> 于2020年10月12日周一 下午11:38写道： > > > Hi， > > 你是想要自己做一个产品，将图显示在Web上？我们是只拿 DAG 中 json 值，然后前端进行处理的。 > > 希望能帮助到你～ > > > > > > Best， > > Hailong Wang > > 在 2020-10-12 18:15:36，"丁浩浩" <18579099...@163.com> 写道： > > >我想要获取到flink webUI上面的DAG图，有什么办法能够获取到吗？ > > >
Re: 当kafka有大量数据积压并且flink冷启动，flink端读取kafka有没有限流的参数？
Hi, 据我所知没有限流的功能，最简单的是设置下任务的并行度 me 于 2020年9月27日周日 下午5:45写道： > flink版本1.11 > flink连接kafka使用的是 flink addSource特性 > > > 原始邮件 > 发件人: me > 收件人: user-zh > 发送时间: 2020年9月27日(周日) 17:22 > 主题: 当kafka有大量数据积压并且flink冷启动，flink端读取kafka有没有限流的参数？ > > > 当kafka有大量数据积压并且flink冷启动，flink端读取kafka有没有限流的参数？ > kafka积压了2亿数据，flink启动后接着之前的状态读取，怎么控制flink的限流？
Re: 怎么样在Flink中使用java代码提交job到yarn
直接扣命令行里提交任务的代码, https://ci.apache.org/projects/flink/flink-docs-stable/ops/cli.html, 这边都是 java 实现的, 转到 spring boot 没啥难度 xiao cai 于2020年9月25日周五 下午5:28写道： > Hi zilong: > > 这种方式我考虑过，个人认为平台层面如果有业务逻辑的侵入，会影响后续的升级。所以我们是在标注输出中正则匹配出jobId和applicationId。你了解YarnClusterDescripto吗？之前社区看到有人用这个提交的。 > > > 原始邮件 > 发件人: zilong xiao > 收件人: user-zh > 发送时间: 2020年9月25日(周五) 17:12 > 主题: Re: 怎么样在Flink中使用java代码提交job到yarn > > > 你提交的任务是可以指定job name的呀，你的job name和你的业务主键绑定就可以做到唯一了，然后根据这个关系查询即可，没记错-ynm > 是指定job name的 xiao cai 于2020年9月25日周五 下午5:01写道： > hi > zilong: > 通过process提交任务以后，通过rest > > api，如何知道哪一个是我提交的呢？如果这时有多个请求过来同时都提交了任务，怎么知道rest返回的application应该与哪一次提交对应呢？ > > > > 原始邮件 > 发件人: zilong xiao > 收件人: user-zh< > user-zh@flink.apache.org> > 发送时间: 2020年9月25日(周五) 16:55 > 主题: Re: > 怎么样在Flink中使用java代码提交job到yarn > > > 我知道呀，你不是从输出里抠的嘛？你想要更优雅的方式可以用rest > api啊，这些信息都是可以拿到的 xiao cai < > flin...@163.com> 于2020年9月25日周五 下午4:53写道： > > hi zilong： > > 你说的这些信息我目前都能拿到，已经实现了。只是实现的方式很不优雅，而且提交的速度很慢。 > > > 原始邮件 > > 发件人: zilong xiao< > acidzz...@gmail.com> > 收件人: user-zh< > user-zh@flink.apache.org> > 发送时间: > 2020年9月25日(周五) 16:48 > 主题: Re: > 怎么样在Flink中使用java代码提交job到yarn > > > > JobId是说Flink的JobId还是yarn上面的application > ID呢？containerID可以通过yarn rest api拿到呀 > > xiao cai > 于2020年9月25日周五 下午4:43写道： > > > > > 使用process没办法拿到当前提交任务的JobId和提交到Yarn上的对应的containerId，自能从标准输出中使用正则解析出来。我们目前使用的就是这种方式，已经实现了。但是这样很不优雅，而且提交的速度也会比较慢。 > > > > > > 原始邮件 > 发件人: zilong xiao > 收件人: user-zh< > > > user-zh@flink.apache.org> > 发送时间: 2020年9月25日(周五) 16:32 > 主题: Re: > > > 怎么样在Flink中使用java代码提交job到yarn > > > > > > Java程序用process调用脚本提交任务没啥问题吧，获取jobId的问题，我理解可以用yarn rest api & > > > > flink rest api来完成，希望对你有帮助，祝好~ xiao cai > 于2020年9月25日周五 > > > 下午4:23写道： > Hi all： > 大家好，我目前遇到一个flink 任务提交方面的困扰： > > > > > 想要在自己的项目中（springboot）提交flink > > > > > job到yarn上。目前采用的方式是process类在本地起进程，调用bin/flink来启动。这样不是很优雅，而且比较难获取到提交的jobId和containerId。看到网上有博客使用ClusterClient的方式来提交，但是遇到了classpath的问题，会缺失一些FLINK_HOME/lib下的jar包 > > > > > 所以我想问，应该如何在自己的应用中提交任务到yarn，既能拿到任务信息，又可以解决classpath为指定带来的困扰。 > 非常感谢 > > > > > > > best, > xiao
Re: 消费kafka数据乱序问题
两个方法 1. kafka 里面可以 keyby, partition 里面都是有序的, 所以每个用户处理都是有序的 2. 就是你说的在 flink 里面做乱序处理 宁吉浩 于2020年9月4日周五 下午5:56写道： > 我也遇到了和你一样的问题，也是两条数据有因果关系，必须有严格的先后顺序，我这边的业务不像银行那么严格； > 我的解决办法是把迟到数据丢弃，然后进行业务计算； > 另起一个程序把数据缓存在内存里，对数据排序，然后再度修正计算； > 之前的实时+离线数仓用的办法，代码开发一次，但还要跑两次； > > > -- > 发件人：smq <374060...@qq.com> > 发送时间：2020年9月4日(星期五) 17:35 > 收件人：wwj ; user-zh > 主 题：回复：消费kafka数据乱序问题 > > > 换句话说，写进kafka的数据是同一个用户的两条，余额分别是0和100，就是我之前那个计算过程，这个计算操作在Oracle完成，我只负责把余额，也就是0或者100更新到kudu对应的余额字段，因为消费数据可能会乱序，先更新100,再更新0,这样导致存在kudu的数据是0,正确的数据最终应该是100 > > ---原始邮件--- > 发件人: "wwj" 发送时间: 2020年9月4日(周五) 下午5:10 > 收件人: "smq"<374060...@qq.com; > 主题: 回复：消费kafka数据乱序问题 > > > > “假如说先处理了存钱，存上之后余额是100-100+100=100” 这句话没看懂，存上之后余额不应该是 100+100=200 吗？ > > > > > > > > > 原始邮件 > > > 发件人："smq"< 374060...@qq.com ; > > 发件时间：2020/9/4 16:40 > > 收件人："user-zh"< user-zh@flink.apache.org ; > > 主题：消费kafka数据乱序问题 > > > > 大家好 > > 现在碰到一个关于处理乱序的问题，业务场景是银行余额的更新，数据源是kafka，有一个账户余额字段，sink到kudu，更新客户余额. > > 如果对于同一个账户的多笔操作出现乱序，可能会导致客户余额不对。比如一个客户账户有100块，先消费100，然后存100，按照正常的数据处理顺序，帐户余额应该还是100的，假如说先处理了存钱，存上之后余额是100-100+100=100,然后处理消费100的那条数据，由于消费在前，这时这条数据对应的余额是100-100=0。 > 这样的话就出现了统计余额错误，请问想要按照事件时间处理数据是不是可以先keyby,然后用watermark.
Re: flink prometheus 无法上报accumulator类型监控吗
accumulator 是聚合后的指标, metics 里是底层的指标, 据我所知没有办法打到监控里面 Yun Tang 于2020年8月28日周五 下午2:37写道： > Hi > > 没有名为 accumulator 的metrics类型数据，目前只有Counters, Gauges, Histograms 和 Meters > [1] 这四种，如果你想要用累积型metrics，可以考虑counters > > [1] > https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/metrics.html#metric-types > > 祝好 > 唐云 > > > From: 赵一旦 > Sent: Friday, August 28, 2020 10:53 > To: user-zh@flink.apache.org > Subject: Re: flink prometheus 无法上报accumulator类型监控吗 > > hi，有人回答下这个问题吗。 > > 赵一旦 于2020年8月21日周五 下午4:20写道： > > > 如题，没找到accumulator类型数据，metric之类找到了，但是accumulator类没找到。 > > >
