Re: Semantic when table joins table from window

Hi Hery, As for choise1: - The state size of join depends on it's input table size, not the result table, so the state size of join of choise1 depends on how many article id, praise id and response_id. - Also non-window join will merge same rows in it's state, i.e, , so the state

Re: getting error while running flink job on yarn cluster

Hi Yubraj, The solution to a similar problem from StackOverflow is to explicitly define the serialVersionUID in your class. For more information, please visit here.[1] [1]: https://stackoverflow.com/questions/27647992/how-resolve-java-io-invalidclassexception-local-class-incompatible-stream-clas

Re: classloading strangeness with Avro in Flink

Hi Cliff, You are welcome, I am very happy to hear this message. Thanks, vino. Cliff Resnick 于2018年8月21日周二 下午11:46写道： > Solved this by moving flink-avro to lib and reverting to > `classloader.resolve-order: parent-first`. I still don't know why, but I > guess if you're reading Avro both from

Some questions about the StreamingFileSink

Hi, I have some questions about the new StreamingFileSink in 1.6. My usecase is pretty simple. I have a cassandra table with 150Millions of lines. They are partitioned by buckets of 100 000 lines. My job is to export each "bucket" to a file (1 bucket = 1 file), so the job is degined like this:

error running flink job on yarn cluster

hi all , i am getting this error java.io.InvalidClassException: com.fasterxml.jackson.databind.ObjectMapper; local class incompatible: stream classdesc serialVersionUID = 1, local class serialVersionUID = 2 at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:616) at

Question about QueryableState

Hi! I’ve started to deploy a small Flink cluster (4tm and 1jm for now on 1.6.0), and deployed a small job on it. Because of the current load, job is completely handled by a single tm. I’ve created a small proxy that is using QueryableStateClient

Re: getting error while running flink job on yarn cluster

it works , but now i am getting org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot instantiate user function. at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperator(StreamConfig.java:239) at

Re: getting error while running flink job on yarn cluster

Hi Yubraj Singh, Can you try submitting with HADOOP_CLASSPATH=`hadoop classpath` set? [1] For example: HADOOP_CLASSPATH=`hadoop classpath` bin/flink run [...] Best, Gary [1]

Re: getting error while running flink job on yarn cluster

It's stage env for the company and other jobs are running on the same so I can't change anything Thanks Yubraj Singh On Tue, Aug 21, 2018, 9:31 PM Paul Lam wrote: > Hi yuvraj, > > Please try turning off timeline server in yarn-site.xml. Currently Flink > does not ship the required dependencies

Re: getting error while running flink job on yarn cluster

Hi yuvraj, Please try turning off timeline server in yarn-site.xml. Currently Flink does not ship the required dependencies for timeline server, which I think could be a bug. Best Regards, Paul Lam > 在 2018年8月21日，22:23，yuvraj singh <19yuvrajsing...@gmail.com> 写道： > > Hi , > > i am getting

Re: classloading strangeness with Avro in Flink

Solved this by moving flink-avro to lib and reverting to `classloader.resolve-order: parent-first`. I still don't know why, but I guess if you're reading Avro both from file and Kafka in the same pipeline then inverted class loader delegation will not work. Thanks, Vino for your help! On Tue,

Re: Unable to start Flink HA cluster with Zookeeper

Yeah, you are right. I have already tried to set up jobmanager.rpc.adress and it works in that case, but if I use this setting I will not be able to use HA, am i right ? How the job manager can register to zookeeper with the right address but not localhost ? -- Sent from:

getting error while running flink job on yarn cluster

Hi , i am getting a error while running a flink job on yarn cluster , its running fine when i run it on flink standalone cluster java.lang.NoClassDefFoundError: com/sun/jersey/core/util/FeaturesAndProperties at java.lang.ClassLoader.defineClass1(Native Method) at

Re: [ANNOUNCE] Apache Flink 1.5.3 released

Great news. Thanks a lot for managing the release Chesnay and to all who have contributed to this release. Cheers, Till On Tue, Aug 21, 2018 at 2:12 PM Chesnay Schepler wrote: > |The Apache Flink community is very happy to announce the release of > Apache Flink 1.5.3, which is the third bugfix

Re: Semantic when table joins table from window

Hi Fabian, So maybe I can not join a table that generate from a window, because the table is getting larger and larger as the time goes, maybe the system will crash one day. I am working on a system that calculate the “score" of article, which is consist of the count of

Re: Unable to start Flink HA cluster with Zookeeper

Hi, In your case the jobmanager binds itself to localhost and that's what it writes to zookeeper. Try starting the jobmanager manually with jobmanager.rpc.address set to the ip of machine you are running the jobmanager. In other words make sure the jobmanager binds itself to the right ip.

Re: Unable to start Flink HA cluster with Zookeeper

FQD or full ip; tried all of them, still no changes ... For ssh connection, I can connect to each machine without passwords. Do you think that the problem can come from : *high-availability.storageDir: file:///shareflink/recovery* ? I don't use a HDFS storage but NAS file system which is

Re: Semantic when table joins table from window

Hi Henry, praiseAggr is an append table, so it contains "100,101,102,100,101,103,100". 1. if you change your sql to s"SELECT article_id FROM praise GROUP BY article_id", the answer is "101,102,103" 2. if you change your sql to s"SELECT last_value(article_id) FROM praise", the answer is "100"

Re: Unable to start Flink HA cluster with Zookeeper

First of all try with FQD or full ip. Also in order to run HA cluster you need to make sure that you have password less ssh access to your slaves and master communication. . On Tue, Aug 21, 2018 at 4:15 PM mozer wrote: > I am trying to install a Flink HA cluster (Zookeeper mode) but the task

Unable to start Flink HA cluster with Zookeeper

I am trying to install a Flink HA cluster (Zookeeper mode) but the task manager cannot find the job manager. Here I give you the architecture; - Machine 1 : Job Manager + Zookeeper - Machine 2 : Task Manager masters: Machine1 slaves : Machine2 flink-conf.yaml:

Unable to start Flink HA cluster with Zookeeper

I am trying to install a Flink HA cluster (Zookeeper mode) but the task manager cannot find the job manager. Here I give you the architecture; - Machine 1 : Job Manager + Zookeeper - Machine 2 : Task Manager masters: Machine1 slaves : Machine2 flink-conf.yaml:

Side effect of DataStreamRel#translateToPlan

Hi all, I noticed that the DataStreamRel#translateToPlan is non-idempotent, and that may cause the execution plan not as what we expected. Every time we call DataStreamRel#translateToPlan (in TableEnvirnment#explain, TableEnvirnment#writeToSink, etc), we add same operators in execution

Re: Semantic when table joins table from window

Hi Fabian, Thanks for your response. This question puzzles me for quite a long time. If the praiseAggr has the following value: window-1 100,101,102 window-2100,101,103 window-3100 the last time the article table joins

[ANNOUNCE] Apache Flink 1.5.3 released

|The Apache Flink community is very happy to announce the release of Apache Flink 1.5.3, which is the third bugfix release for the Apache Flink 1.5 series. Apache Flink® is an open-source stream processing framework for distributed, high-performing, always-available, and accurate data

Re: Semantic when table joins table from window

Hi, The semantics of a query do not depend on the way that it is used. praiseAggr is a table that grows by one row per second and article_id. If you use that table in a join, the join will fully materialize the table. This is a special case because the same row is added multiple times, so the

Re: classloading strangeness with Avro in Flink

Hi Aljoscha, We need flink-shaded-hadoop2-uber.jar because there is no hadoop distro on the instance the Flink session/jobs is managed from and the process that launches Flink is not a java process, but execs a process that calls the flink script. -Cliff On Tue, Aug 21, 2018 at 5:11 AM Aljoscha

Re: Will idle state retention trigger retract in dynamic table?

Hi, No, I don't think this behavior is weird. If we would retract when idle state is discarded, the result would no longer correspond to the query. So we would produce incorrect results even if the removed state would never by used again. If you want to have consistent, exact results you need

Re: Will idle state retention trigger retract in dynamic table?

Hi Henry, Idle state retention is just making a trade-off between the accuracy and the storage consumption. It can meet part of the calculation requirements in the stream environment, but not all. For instance, in your use case, if there exists a TTL for each article, their praise states can

Semantic when table joins table from window

Hi All, var praiseAggr = tableEnv.sqlQuery(s"SELECT article_id FROM praise GROUP BY HOP(updated_time, INTERVAL '1' SECOND,INTERVAL '3' MINUTE) , article_id" ) tableEnv.registerTable("praiseAggr", praiseAggr) var finalTable = tableEnv.sqlQuery(s”SELECT 1 FROM article a join

Re: Could not cancel job (with savepoint) "Ask timed out"

Just a small addition. Concurrent cancel call will interfere with the cancel-with-savepoint command and directly cancel the job. So it is better to use the cancel-with-savepoint call in order to take savepoint and then cancel the job automatically. Cheers, Till On Thu, Aug 9, 2018 at 9:53 AM

Re: Will idle state retention trigger retract in dynamic table?

Hi Fabian, Is the behavior a bit weird? Because it leads to data inconsistency. Best, Henry > 在 2018年8月21日，下午5:14，Fabian Hueske 写道： > > Hi, > > In the given example, article_id 123 will always remain in the external > storage. The state is removed and hence it cannot be retracted

Re: 答复: Best way to find the current alive jobmanager with HA mode zookeeper

Hi Martin, when configuring Flink to use the ZooKeeper HA mode, then it won't be necessary to specify the leader's address manually. The CLI will ask ZooKeeper for the leader information and send the request to the current leader. This should work with at least Flink >= 1.5 and also with Flink

Re: Job Manager killed by Kubernetes during recovery

Hi Bruno, in order to debug this problem we would need a bit more information. In particular, the logs of the cluster entrypoint and your K8s deployment specification would be helpful. If you have some memory limits specified these would also be interesting to know. Cheers, Till On Sun, Aug 19,

Re: Will idle state retention trigger retract in dynamic table?

Hi, In the given example, article_id 123 will always remain in the external storage. The state is removed and hence it cannot be retracted anymore. Once the state was removed and the count reaches 10, a second record for article_id 123 will be emitted to the data store. As soon as you enable

Re: classloading strangeness with Avro in Flink

Hi Cliff, Do you actually need the flink-shaded-hadoop2-uber.jar in lib. If you're running on YARN, you should be able to just remove them because with YARN you will have Hadoop in the classpath anyways. Aljoscha > On 21. Aug 2018, at 03:45, vino yang wrote: > > Hi Cliff, > > If so, you

Re: UTF-16 support for TextInputFormat

Thanks for creating FLINK-10134 and adding your suggestions! Best, Fabian 2018-08-13 23:55 GMT+02:00 David Dreyfus : > Hi Fabian, > > I've added FLINK-10134. FLINK-10134 > . I'm not sure you'd > consider it a blocker or that I've identified the

Re: Will idle state retention trigger retract in dynamic table?

Hi Fabian, SELECT article_id FROM praise GROUP BY article_id having count(1)>=10 If article_id 123 has 100 praises and remains its state in the dynamic table ,and when the time passed, its state is removed, but later the article_id 123 has never reached to 10 praises. How

Re: 答复: Best way to find the current alive jobmanager with HA mode zookeeper

Hi guys, Just to close the loop, with the Flink 1.3.2 cli you have to provide the Flink Job Manager host address in order to submit a job like so: ${FLINK_HOME}/bin/flink run -d -m ${FLINK_JOBMANAGER_ADDRESS} ${JOB_JAR} Since we are running the DCOS Flink package we use the Marathon rest api to

Re: Will idle state retention trigger retract in dynamic table?

Hi, No, it won't. I will simply remove state that has not been accessed for the configured time but not change the result. For example, if you have a GROUP BY aggregation and the state for a grouping key is removed, the operator will start a new aggregation if a record with the removed grouping

Will idle state retention trigger retract in dynamic table?

Hi All, Will idle state retention trigger retract in dynamic table? Best, Henry

Re: Flink checkpointing to Google Cloud Storage

Hey, >From my perspective, such issues always meant clashing dependencies in case of Flink. Have you checked the full dependency tree if there are no issues there ? Best Regards, Dominik.

Re: Data loss when restoring from savepoint

I realized that BucketingSink must not play any role in this problem. This is because only when the 24-hour window triggers, BucketinSink gets a burst of input. Around the state restoring point (middle of the day) it doesn't get any input, so it can't lose anything either (right?). I will next

Re: How do I investigate checkpoints failures

Hi Alex, First thing to do in such cases is to analyze logs for jobmanager and taskmanagers and look for exceptions there. The cause for latest failed checkpoint says the checkpoint expired. You can try increasing the checkpoint timeout (you can check more configuration options for checkpoints

How do I investigate checkpoints failures

Hello, I have a cluster with multiple jobs running on it. One of the jobs has checkpoints constantly failing [image: image.png] How do I investigate it? Thank you, Alex

How to pass a dynamic path while writing to files using writeFileAsText(path)?

Let's say I have a Stream with elements of type `String`. I want to write each element in the stream to a separate file in some folder. I'm using the following set up. > filteredStream.writeAsText(path).setParallelism(1); How do I make this path variable? I even tried adding

Re: How does flink know which data is modified in dynamic table?

Hi， You are right. We can make use of it to do soft delete. But there will be problems in other cases. For example, retract messages by the whole row. I opened a jira[1] about this problem. Thanks for bring up this discussion. [1] https://issues.apache.org/jira/browse/FLINK-10188 Best, Hequn