Re: Flink SQL Count Distinct performance optimization

Hi, Could you try to find out what's the bottleneck of your current job? This would leads to different optimizations. Such as whether it's CPU bounded, or you have too big local state thus stuck by too many slow IOs. Best, Kurt On Wed, Jan 8, 2020 at 3:53 PM 贺小令 wrote: > hi sunfulin, > you ca

Re: Flink SQL Count Distinct performance optimization

hi sunfulin, you can try with blink planner (since 1.9 +), which optimizes distinct aggregation. you can also try to enable *table.optimizer.distinct-agg.split.enabled* if the data is skew. best, godfreyhe sunfulin 于2020年1月8日周三 下午3:39写道： > Hi, community, > I'm using Apache Flink SQL to build so

Flink SQL Count Distinct performance optimization

Hi, community, I'm using Apache Flink SQL to build some of my realtime streaming apps. With one scenario I'm trying to count(distinct deviceID) over about 100GB data set in realtime, and aggregate results with sink to ElasticSearch index. I met a severe performance issue when running my flink jo

Re: Using redis cache in flink

Hi Yun, Thanks, the way I want to use redis is like a cache not as state backend. I would still have rocksdb state backend for other states. The reason to use cache instead of managed state is because I’d get around 10k msgs per task slot and I don’t have to get the state from rocksdb for each loo

Re: How to verify if checkpoints are asynchronous or sync

If you want to figure out the performance problem, maybe async-profile[1] can be helpful [1] https://github.com/jvm-profiling-tools/async-profiler Best, Congxian William C 于2020年1月8日周三 上午11:37写道： > Hallo > > on 2020/1/8 11:31, RKandoji wrote: > > I'm running my job on a EC2 instance with 32 cor

Re: Using redis cache in flink

Hi Navneeth If you wrap redis as a state backend, you cannot easily share data across slots as Flink construct state backend per operator with local thread only. If you use a redis cluster as a externalized service to store your data, you can share data across slots easily. However, compared wi

Using redis cache in flink

Hi All, I want to use redis as near far cache to store data which are common across slots i.e. share data across slots. This data is required for processing every single message and it's better to store in a in memory cache backed by redis rather than rocksdb since it has to be serialized for ever

回复：jobgraph 生成

Very sorry for the wrong operation. I copied the wrong email address by the phone. Thank you for your reply. | | 张江 | | 邮箱：zjkingdom2...@163.com | 签名由 网易邮箱大师 定制 在2020年01月08日 11:08，tison 写道： Hi Zhang, I just notice that it is sent to user list. Please send to user-zh list(in cc) next time

Re: managedMemoryInMB failure

Hi Fanbin, The blink planner batch sql operators requires managed memory, and the amount of managed memory needed depends on your job. The failure is because the slot, according to your cluster configurations, does not have enough managed memory to fulfill the requests. To fix the problem, you wo

Re: How to verify if checkpoints are asynchronous or sync

Hallo on 2020/1/8 11:31, RKandoji wrote: I'm running my job on a EC2 instance with 32 cores and according to the documentation I tried to use as many task slots the number of cores, numOfTaskSlots=32 and parallelism=32. But I noticed that the performance is slightly degrading when I'm using 32

Re: How to verify if checkpoints are asynchronous or sync

Sorry for not providing the context earlier. I'm running my job on a EC2 instance with 32 cores and according to the documentation I tried to use as many task slots the number of cores, numOfTaskSlots=32 and parallelism=32. But I noticed that the performance is slightly degrading when I'm using 32

Re: How to verify if checkpoints are asynchronous or sync

Hi RocksDB supports Incremental and full snapshot, both are async, do you want to verify whether it's incremental or full snapshot? but I don't know an easy to get this information currently Best, Congxian Zhijiang 于2020年1月8日周三 上午10:56写道： > The log way is simple for tracing and you can also g

Re: jobgraph 生成

A public way to get JSON plan of a JobGraph is, with an existing Flink Cluster, use REST API JarPlan[1]. Best, tison. [1] https://ci.apache.org/projects/flink/flink-docs-master/monitoring/rest_api.html#jars-jarid-plan tison 于2020年1月8日周三 上午11:08写道： > Hi Zhang, > > I just notice that it is sent

Re: jobgraph 生成

Hi Zhang, I just notice that it is sent to user list. Please send to user-zh list(in cc) next time if you want to discuss in Chinese. Best, tison. tison 于2020年1月8日周三 上午11:06写道： > 如果你是流作业的话，参考这个页面[1]搞到 JobGraph 之后可以 > > JsonPlanGenerator.generatePlan(jobGraph) > > 拿到 JobGraph 的 JSON。不过这个是非常内部的

Re: jobgraph 生成

如果你是流作业的话，参考这个页面[1]搞到 JobGraph 之后可以 JsonPlanGenerator.generatePlan(jobGraph) 拿到 JobGraph 的 JSON。不过这个是非常内部的逻辑，没有任何兼容性保障。 Best, tison. [1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/deployment/mesos.html#flink-job-cluster-on-mesos 张江 于2020年1月8日周三 上午11:01写道： > 大家好， > > 通过调

jobgraph 生成

大家好， 通过调用env.getExecutionPlan可以以json格式得到streamGraph，放到flink visualizer里可以得到可视化结果。现在我想以类似方法得到 jobGraph，生成jobGraph的json文件。不知道应该怎么做？ flink里似乎没有直接的API可以调用，但是我在flink web ui界面上可以看到job经过chain后的执行图，如下图所示。这个是怎么生成的？ 谢谢 | | 张江 | | 邮箱：zjkingdom2...@163.com | 签名由 网易邮箱大师 定制

Re: How to verify if checkpoints are asynchronous or sync

The log way is simple for tracing and you can also grep some keywords to find your requirement messages to avoid skimming through the whole large logs. I am not quite sure what's your specific motivation for doing this. Besides the log way, you can also monitor the thread stack for confirming whe

Re: How to verify if checkpoints are asynchronous or sync

Thanks for the reply. I will check and enable debug logs specifically for the class that contains this log. But in general logs are already too huge and I'm trying to suppress some of them, so wondering if there is any other way? Thanks, RKandoji On Tue, Jan 7, 2020 at 7:50 PM William C wrote:

Re: How to verify if checkpoints are asynchronous or sync

Can you enable debug log to check with that? regards. on 2020/1/8 6:36, RKandoji wrote: But I'm curious if there is way to verify if the checkpoints are happening asynchronously or synchronously.

managedMemoryInMB failure

Hi, with Flink 1.9 running in docker mode, I have a batch job and got the following error message. However, it works totally fine with the same code on EMR. I checked the log and here is the only difference: managedMemoryInMB=138 . (the working ones has 0 value) did anybody see this before? Than

Re: Duplicate tasks for the same query

hi Kurt, Thanks for the additional info. RK On Sun, Jan 5, 2020 at 8:33 PM Kurt Young wrote: > Another common skew case we've seen is null handling, the value of the > join key > is NULL. We will shuffle the NULL value into one task even if the join > condition > won't stand by definition. > >

How to verify if checkpoints are asynchronous or sync

Hi, I'm using Flink 1.9, BlinkPlanner and rocksDB for backend with checkpointing enabled. I understand that checkpointing is async by default for RocksDB. But I'm curious if there is way to verify if the checkpoints are happening asynchronously or synchronously. Please let me know. Thanks, RKan

Re: Abstract classes in Stateful functions

Thanks! Den tis 7 jan. 2020 12:49Igal Shilman skrev: > Hi Dan, > Yes. You should be able to store “wheels” in “Vehicle”. > > Igal. > > On Monday, January 6, 2020, Dan Pettersson > wrote: > >> Hello again :-) >> >> When using an abstract class should the instance variables be stored in >> Persis

Re: Flink logging issue with logback

A quick update. The suppression of stdout/stderr actually might soon be dropped, see: https://issues.apache.org/jira/browse/FLINK-15504 Best, Dawid On 07/01/2020 07:17, Yang Wang wrote: > Hi Bajaj, > > I have tested just as you say, and find that the logs in the user > class could not show up wh

Re: Flink logging issue with logback

Hi Bajaj, I spent a bit more time on that and I figured out the difference. The cause is in the implementation of log4j's ConsoleAppender. So first of all as I said before, the code path that submits a job through Web UI redirects the std.out and std.err to a memory buffer for the time of the "ma

Re: Operators resource requirements on K8s Flink session cluster

Hi Yang, Thanks for your quick reply. The Flink K8s documentation distinguishes between standalone and session deployment mode. I think I will use the latter. Since my previous mail, I found FLIP-53

Abstract classes in Stateful functions

Hi Dan, Yes. You should be able to store “wheels” in “Vehicle”. Igal. On Monday, January 6, 2020, Dan Pettersson wrote: > Hello again :-) > > When using an abstract class should the instance variables be stored in > PersistenceValues > to conform to the fault tolerance? > > Exempel Car, Truck a