Re: Pre shuffle aggregation in flink is not working

Hi Jing, I tried using `*MapBundleOperator*` also (I am yet to test with LinkedHashMap) . But I am always seeing that the following code of ` *AbstractMapBundleOperator.java*` `*numOfElements` *is always 0. It is never getting incremented. I replaced `*TaxiFareStream*` with ` *MapBundleOperator*`

Re: Kafka Metrics

FYI, I'm referring to the legacy offsets metric gauges. On Thu, Aug 19, 2021 at 4:53 PM Mason Chen wrote: > Hi all, > > We have found that the per partition Kafka metrics contributes to a lot of > metrics being indexed by our metrics system. > > We would still like to have the proxied kafka metr

submit new job is not working

hello all ,        I was able to run sample example and was able to upload jar using UI, cluster which I have deployed on k8s  Today I had to reboot jobmanager after that I am not able to upload any jar to my cluster. Do not see any log as well to debug  any help  --kumar 

Re: Error while deserializing the element

Hi Vijay, Yun, I've created a JIRA https://issues.apache.org/jira/browse/FLINK-23886 to track this. Best, JING ZHANG JING ZHANG 于2021年8月20日周五 下午1:19写道： > Hi Vijay, > I have encountered the same problem several times in online production > Flink jobs, but I have not found the root cause of the e

Re: Error while deserializing the element

Hi Vijay, I have encountered the same problem several times in online production Flink jobs, but I have not found the root cause of the exception yet. We have walk around the exception by adding the following parameter, hope it could help you. state.backend.rocksdb.timer-service.factory: HEAP I wo

Re: DataStream to Table API

Hi! I've created a JIRA ticket[1] for this issue. Please check it out and track the progress there. [1] https://issues.apache.org/jira/browse/FLINK-23885 Caizhi Weng 于2021年8月20日周五 上午10:47写道： > Hi! > > This is because TypeExtractor#getMapReturnTypes are not dealing with row > types (see that me

Re: Periodic output at end of stream

Hi Matthias, Thanks for providing the example, I would reply back soon after I do some debug. Best, JING ZHANG Matthias Broecheler 于2021年8月19日周四 上午1:53写道： > Hey JING, > > thanks for getting back to me. I tried to produce the smallest, > self-contained example that produces the phenomenon: > htt

Re: Pre shuffle aggregation in flink is not working

Hi Suman, Please try copy `*MapBundleOperator*`, update the `HashMap` to `LinkedHashMap` to keep the output sequence consistent with input sequence. Best, JING ZHANG suman shil 于2021年8月20日周五 上午2:23写道： > Hi Jing, > Thanks for looking into this. Here is the code of `TaxiFareStream'. I was > follo

Re: How can I achieve 'sink.partition-commit.policy.kind'='metastore,success-file' with batch Hive sink?

An example result directory on HDFS looks like this: >> >> [10.106.11.21:serv@cn-hz-wl-prod-data-stat00:~]$ hdfs dfs -ls >> /user/hive/warehouse/aiinfra.db/user_loss_predictions/p_day=20210819 >> Found 9 items >> -rw-r- 2 basedata aiinfra 0 2021-08-20

Re: How can I achieve 'sink.partition-commit.policy.kind'='metastore,success-file' with batch Hive sink?

6.11.21:serv@cn-hz-wl-prod-data-stat00:~]$ hdfs dfs -ls > /user/hive/warehouse/aiinfra.db/user_loss_predictions/p_day=20210819 > Found 9 items > -rw-r- 2 basedata aiinfra 0 2021-08-20 08:56 > /user/hive/warehouse/aiinfra.db/user_loss_predictions/p_day=20210819/_SUCCESS > -r

Re: DataStream to Table API

Hi! This is because TypeExtractor#getMapReturnTypes are not dealing with row types (see that method and also TypeExtractor#privateGetForClass). You might want to open a JIRA ticket for this. Matthias Broecheler 于2021年8月20日周五 上午7:01写道： > Hey Flinkers, > > I am trying to follow the docs >

How can I achieve 'sink.partition-commit.policy.kind'='metastore,success-file' with batch Hive sink?

/warehouse/aiinfra.db/user_loss_predictions/p_day=20210819 Found 9 items -rw-r- 2 basedata aiinfra 0 2021-08-20 08:56 /user/hive/warehouse/aiinfra.db/user_loss_predictions/p_day=20210819/_SUCCESS -rw-r- 2 basedata aiinfra 10684668 2021-08-20 08:49 /user/hive/warehouse/

Kafka Metrics

Hi all, We have found that the per partition Kafka metrics contributes to a lot of metrics being indexed by our metrics system. We would still like to have the proxied kafka metrics from the kafka clients library. Is there a flag to only exclude Flink's additional Kafka metrics? Best, Mason

Looking for suggestions about multithreaded CEP to be used with flink

Hi, Here's our use case : We are planning to build a rule based engine on top of flink with huge number of rules(1000s). the rules could be stateless or stateful. Example stateless rule is : A.id = 3 && A.name = 'abc' || A.color = red. Example stateful rule is : A is event.id =3, B is event.name

DataStream to Table API

Hey Flinkers, I am trying to follow the docs to convert a DataStream to a Table. Specifically, I have a DataStream of Row and want the columns of the row to become the columns of the resulting table. Tha

RE: failures during job start

Thanks you. I am going to try the first option for now, but I do need to figure out why deployment takes so long. Are there any metrics or log patterns that would indicate which task is waiting and which task is being waited on? From: Chesnay Schepler Sent: Thursday, August 19, 2021 2:23 PM

Re: Process suspend when get Hana connection in open method of sink function

If the Hana driver cannot be loaded then the most likely reason is that the dependency is not actually on the classpath. Please double-check that your user jar bundles the dependency. On 18/08/2021 15:05, Chenzhiyuan(HR) wrote: Dear all: I have a problem when I want to sink data to Hana datab

Re: Pre shuffle aggregation in flink is not working

Hi Jing, Thanks for looking into this. Here is the code of `TaxiFareStream'. I was following this link http://felipeogutierrez.blogspot.com/2019/04/implementing-my-own-stream-operator-in.html . Please let me know if there is any other way of aggregating elements locally. *public class

Re: failures during job start

This exception means that a task was deployed, but the task that produces the data it wants to consume was not available yet (even after waiting for a while). Your case sounds similar to https://issues.apache.org/jira/browse/FLINK-9413, where this happens because the deployment of the produci

Re: Just failed while starting

Can you share the logs with us (ideally on DEBUG if available) from the affected TaskManager and JobManager? On 19/08/2021 08:29, Ivan Yang wrote: Dear Flink community, I recently running into this issue at a job startup. It happened from time to time. Here is the exception from the job manag

Re: Job manager sometimes doesn't restore job from checkpoint post TaskManager failure

How do you deploy Flink on Kubernetes? Do you use the standalone or native

Re: map concurrent modification exception analysis when checkpoint

Essentially this exception means that the state was modified while a snapshot was being taken. We usually see this when users hold on to some state value beyond a single call to a user-defined function, particularly from different threads. We may be able to pinpoint the issue if you were to p

Re: Metrics outside of RichFunction context

I don't believe there are other options, outside of creating a ticket to have Flink extend the API accordingly. On 19/08/2021 16:40, John Karp wrote: Hi, I'm using StreamingFileSink to collect records into avro files. Inside of the BulkWriter implementation, I have to do some operations (such

Re: Apache Flink matrics are not alligned in the reporter

What reporter interval do you have configured? On 19/08/2021 13:31, Jawad Tahir wrote: Hi, I have defined Graphite as my matrics reporter with my Flink (v1.13.2). My pipeline is pretty simple. It consists of one source, one stateful operator (simple window aggregation), and one sink (operati

Re: Pre shuffle aggregation in flink is not working

Hi Suman, Would you please provide the code about `*TaxiFareStream*`? It seems we could use `MapBundleOperator` directly here. BTW, I have some concerns about using the solution to do local-aggregation for window aggregation because `MapBundleOperator` would save input data in a bundle which is a H

map concurrent modification exception analysis when checkpoint

Flink web ui shows the exception as follows. In the task (ual_transform_UserLogBlackUidJudger -> ual_transform_IpLabel ), the first one is a broadcast process function, and the second one is an async function. I do not know whether the issues have some relation to it. And the issues not occurred b

Metrics outside of RichFunction context

Hi, I'm using StreamingFileSink to collect records into avro files. Inside of the BulkWriter implementation, I have to do some operations (such as dynamic schema lookup) which I want to record metrics about. However, the BulkWriter API, as it is defined, does not accept a RuntimeContext or Metrics

Job manager sometimes doesn't restore job from checkpoint post TaskManager failure

Hi all, I've noticed that sometimes when task managers go down--it looks like the job is not restored from checkpoint, but instead restarted from a fresh state (when I go to the job's checkpoint tab in the UI, I don't see the restore, and the number in the job overview all get reset). Under what c

Re: How can I build the flink docker image from source code?

Hi! If you only modified Java code, use mvn clean package to build Flink from source code. After that COPY all jars in flink-dist/target/flink-/lib to the lib directory of the latest Flink image. Chenyu Zheng 于2021年8月19日周四 下午7:36写道： > Hi contributors, > > > > I’ve changed a little bit code in f

How can I build the flink docker image from source code?

Hi contributors, I’ve changed a little bit code in flink, and want to build a docker image to test it. Could you tell me how can I build the image from source code? Thx!

Apache Flink matrics are not alligned in the reporter

Hi, I have defined Graphite as my matrics reporter with my Flink (v1.13.2). My pipeline is pretty simple. It consists of one source, one stateful operator (simple window aggregation), and one sink (operations-playground, basically). I have set the parallel factor as 2. The graph of the pipeline is

Re: Theory question on process_continously processing mode and watermarks

I think what you are seeing is that the files have records with similar timestamps. That means after reading file1 your watermarks are already progressed to the end of your time range. When Flink picks up file2, all records are considered late records and no windows fire anymore. See [1] for a pos

Re: Theory question on process_continously processing mode and watermarks

Hi! FileProcessingMode.PROCESS_CONTINUOUSLY means to continuously scans the file for updates, and there should be nothing to do with stopping the streaming job. I'm suspecting that in the column you defined the watermark there is some data which exceeds Long.MAX_VALUE. A Long.MAX_VALUE watermark

Re: Timer Service vs Custom Triggers

Hi! If you'd like to aggregate something on the records before time out, then you want to consider using session window (instead of writing your own trigger). However if aggregation is not needed I would prefer using a process function to process watermark by myself, as the registered timer in the

Theory question on process_continously processing mode and watermarks

Hello, during my personal development of a Flink streaming Platform i found something that perplexes me.Using FileProcessingMode.PROCESS_CONTINUOUSLYInto a streaming job that uses tumbling Windows and watermarks causes my streaming process to stop ad the reading files phase.Meanwhile if i delete my

Re: Setting S3 parameters in a K8 jobmanager deployment

I am afraid jobmanager.sh[1] could not parse the "-D" correctly now. [1]. https://github.com/apache/flink/blob/master/flink-dist/src/main/flink-bin/bin/jobmanager.sh Best, Yang Robert Cullen 于2021年8月18日周三 下午10:21写道： > I have a kubernetes jobmanager deployment that requires parameters be > pas