Re: Convert BIGINT to TIMESTAMP in pyflink when using datastream api

Hi Shilpa, There might be something wrong when defining the rowtime field with the Connector descriptor, it’s recommended to use SQL DDL to create tables, and do queries with table API. Best, Shuiqiang Shilpa Shankar 于2021年3月4日周四 下午9:29写道： > Hello, > > We are using pyflink's datastream api v1.

Re: how to propagate watermarks across multiple jobs

Thank you. Yuan Mei 于2021年3月4日周四 下午11:10写道： > Hey Yidan, > > KafkaShuffle is initially motivated to support shuffle data > materialization on Kafka, and started with a limited version supporting > hash-partition only. Watermark is maintained and forwarded as part of > shuffle data. So you are ri

Re: Allocating tasks to specific TaskManagers

Are trying to do something along the lines of "deploy this subtask to a TM that has a GPU"? If so, then the community is currently working on supporting such use-cases; see FLIP-56 / FLIP-156

Re: Allocating tasks to specific TaskManagers

re-adding user mailing list pt., 5 mar 2021 o 01:45 황혜조 napisał(a): > Hi Piotrek, > > I am evaluating flink for constraint aware processing, which needs > specific tasks being assigned to specific task managers. > > For that, I passed registeredTaskManagers from JobMaster. However when I > check

Multiple JobManager HA set up for Standalone Kubernetes

I am trying to figure out the right architecture for running Standalone Kubernetes with Job manager HA. The documentation for running HA seems to always suggest that there needs to be multiple job managers, but there isn'

Re: Job downgrade

Hi Gordon, I was using RocksDB backend Alexey From: Tzu-Li (Gordon) Tai Sent: Thursday, March 4, 2021 12:58:01 AM To: Alexey Trenikhun Cc: Piotr Nowojski ; Flink User Mail List Subject: Re: Job downgrade Hi Alexey, Are you using the heap backend? If that's th

Re: [Flink-SQL] FLOORing OR CEILing a DATE or TIMESTAMP to WEEK uses Thursdays as week start

Post the created Jira link https://issues.apache.org/jira/browse/FLINK-21624 Best, Leonard > > Thanks a lot for the added context and pointers Julian and Leonard, > > I've fixed it by going down to the arithmetics as suggested in one of the

[DISCUSSION] Introduce a separated memory pool for the TM merge shuffle

Hi, all In the Flink 1.12 we introduce the TM merge shuffle. But the out-of-the-box experience of using TM merge shuffle is not very good. The main reason is that the default configuration always makes users encounter OOM [1]. So we hope to introduce a managed memory pool for TM merge shuffle to

Re: [EXTERNAL] Re: 1.12.2 docker image

It seems that some tags in the official Flink image repository do not update to 1.12.2. Could you please have a try on the "apache/flink" repository? Best, Yang Bohinski, Kevin 于2021年3月5日周五 上午6:25写道： > Hi, > > > > Actually that seemed to use a cached 1.12.1 image. > > > > I’m seeing the followi

Re: Python Flink cluster: how to shut-down

I think chesnay's answer is on the point. You could find how to list and cancel a Flink application on Kubernetes here[1]. Another thing is that please make sure that you are using the correct service exposed type[2]( e.g. LoadBalancer on the cloud, NodePort for the self-managed cluster, or Cluste

Re: PyFlink Connection Refused to Kubernetes Session Cluster

Hi Robert, It seems the retrieved address of JobManager is a cluster-internal Ip that can noly be accessed inside the cluster. As you said, you might need to create an ingress to expose the JobManager service so that the client can access to it outside of the k8s cluster. Best, Shuiqiang Robert

Re: Watermark doesn't progress after job restore from savepoint

Hi, I think you could try implementing the `CheckpointedFunction` interface and `FunctionInitializationContext.isRestored` is an indicator for that. BTW: I am not very sure your scenarios but maybe you could try to set idleness configurations [1] [1] https://ci.apache.org/projects/flink/flink-doc

Re: Flink application kept restarting

Thanks Julian. Appreciated your help. I will check task manager logs and decide pick which approah. Best regards Rainie On Thu, Mar 4, 2021 at 3:04 PM Jaffe, Julian wrote: > Hey Rainie, > > > > Kafka internally attempts to retry topic metadata fetches if possible. If > you think the root cause

Re: Running StreamingJob.java in apache flink workflow

The quickstarts provide you with a skeleton for building applications. It provides you with pre-defined entrypoints (like StreamingJob) and the recommended maven setup to package this application into a jar. However, they do not provide a running application. When you look at the StreamingJob

Re: Flink application kept restarting

Hey Rainie, Kafka internally attempts to retry topic metadata fetches if possible. If you think the root cause was just due to network congestion or the like, you might want to look into increasing `request.timeout.ms`. Because of the internal retry attempts, however, this exception usually mea

Re: [EXTERNAL] Re: 1.12.2 docker image

Hi, Actually that seemed to use a cached 1.12.1 image. I’m seeing the following: ``` ➜ docker pull flink:1.12.2 1.12.2: Pulling from library/flink no matching manifest for linux/amd64 in the manifest list entries ``` Best kevin From: "Bohinski, Kevin" Date: Thursday, March 4, 2021 at 4:17 PM T

Re: Flink application kept restarting

Hi Matthias, Do you have any suggestions to handle timeout issues when fetching data from a Kafka topic? I am thinking of adding a retry logic into flink job, not sure if this is the right direction. Thanks again Best regards Rainie On Wed, Mar 3, 2021 at 12:24 AM Matthias Pohl wrote: > Hi Rai

Re: [EXTERNAL] Re: 1.12.2 docker image

Hi, I see the images on docker hub. I tried to launch a k8s session with the following options, and I am getting the following logs: ``` -Dkubernetes.container.image=flink:1.12-scala_2.12-java8 \ -Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.12.2.jar \ -Dcont

Re: pipeline.auto-watermark-interval vs setAutoWatermarkInterval

Correction: The first link was supposed to be, 1. pipeline.auto-watermark-interval https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/config.html#pipeline-auto-watermark-interval On Wed, Mar 3, 2021 at 7:46 PM Aeden Jameson wrote: > > I'm hoping to have my confusion clarifie

Re: Debugging long Flink checkpoint durations

The checkpoint was only acknowledged shortly after it was started. On Thu, Mar 4, 2021 at 12:38 PM Dan Hill wrote: > I dove deeper into it and made a little more progress (by giving more > resources). > > Here is a screenshot of one bottleneck: > https://drive.google.com/file/d/1CIatEuIJwmKjBE9_

Re: Debugging long Flink checkpoint durations

I dove deeper into it and made a little more progress (by giving more resources). Here is a screenshot of one bottleneck: https://drive.google.com/file/d/1CIatEuIJwmKjBE9__RihVlxSilchtKS1/view My job isn't making any progress. It's checkpointing and failing. The taskmaster text logs are empty d

PyFlink Connection Refused to Kubernetes Session Cluster

Attempting to run the word_count.py example on my kubernetes (session) cluster: ./bin/flink run \ --target kubernetes-session \ -Dkubernetes.cluster-id=cmdaa \ -Dkubernetes.container.image=cmdaa/pyflink:0.0.1 \ --pyModule word_count \ --pyFiles /opt/flink-1.12.2/examples/python/table/batch/word_co

Re: Watermark doesn't progress after job restore from savepoint

Thanks Piotr. Got it. Had to push the static rules to the kafka queue as it had expired and got archived from the topic. Post this the pipeline resumed. To your suggestion on implementing an operator that remembers the watermark, is there any indicator that the job has been resumed which I can use

Re: Compile Error

Could you paste the contents of the properties section of your pom.xml? target.java.version should be set to 1.8; I do not understand why it tries to compile to another java version. On 3/4/2021 5:24 PM, Abdullah bin Omar wrote: I used mvn install. it shows this one: SLF4J: Actual binding is

Re: reading file from s3

Sure, This is the full exception stacktrace: Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed. at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144) at org.apache.flink.runtime.minicluster.MiniClusterJobClient

AW: State Schema Evolution within SQL API

I think simply changing the parallelism or upgrading the underlying Flink version in a cluster should be no problem. But you want to change your queries, right ? Best, Jan Von: Jan Oelschlegel Gesendet: Donnerstag, 4. März 2021 12:26 An: XU Qinghui ; Dawid Wysakowicz Cc: user Betreff: AW: S

Re: Compile Error

I used mvn install. it shows this one: SLF4J: Actual binding is of type [org.slf4j.impl.SimpleLoggerFactory] SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/Users/abdullah/.p2/pool/plugins/org.eclipse.m2e.maven.runtime.slf4j.simple_1.16.0.20200610-1735/jars

Re: reading file from s3

Can you show us the full exception stacktrace? Intuitively I would think your cluster configuration contains an invalid value for some memory configuration option. On 3/4/2021 4:45 PM, Avi Levi wrote: Hi , I am pretty new. I am keep on struggling to read a file from s3 but getting this weird 

Re: Compile Error

I think your confusion is caused by using using "mvn clean", instead of "mvn clean package". "mvn clean" cleans up, deleting jars and such, whereas "mvn package" compiles the code and packages a jar. On 3/4/2021 3:41 PM, Abdullah bin Omar wrote: Hi Chesnay, Thank you for your reply. Sorry f

reading file from s3

Hi , I am pretty new. I am keep on struggling to read a file from s3 but getting this weird exception : Caused by: java.lang.NumberFormatException: For input string: "64M" (if anyone can link me to a working github example that will be awesome) . what am i doing wrong? This is how my code looks li

Re: Scaling Higher than 10k Nodes

Hi Joey, We are currently running around 2000+ small Flink clusters on top of k8s, currently at around ~ 100 nodes. Do you see yourself scaling to 10k nodes, given that each node can run a significant amount of Flink jobs inside of it? On Thu, Mar 4, 2021 at 10:51 AM Piotr Nowojski wrote: > May

Re: Python Flink cluster: how to shut-down

run-application creates an application cluster that shuts down once the job is complete. As such, canceling the job in this case is equivalent to shutting down the cluster. AFAIK you also need to specify kubernetes arguments when using the list command. (without any argument it just assumes yo

Python Flink cluster: how to shut-down

I ran this command using the example from the CLI page on the Flink website: $ ./bin/flink run-application \ —target kubernetes-application \ —parallelism 8 \ -Dkubernetes.cluster-id= \ -Dtaskmanager.memory.process.size=4096m \ -Dkubernetes.taskmanager.cpu=2 \ -Dtaskmanager.numberOfTaskSlots=4 \ -

Re: Watermark doesn't progress after job restore from savepoint

Hi Hemant, State of the latest seen watermarks is not persisted in the operators. Currently DataStream API assumes that after recovery watermarks are going to be re-emitted sooner or later. What probably happens is that one of your sources has emitted watermarks (maybe some very high one or even `

Re: Defining GlobalJobParameters in Flink Unit Testing Harnesses

hmm...at a glance I don't anything else that would prevent things from working. I couldn't find a place where we copy the ExecutionConfig, so you should be modifying the very config that is passed to your function. It seems we inevitably have to dive deep into the source / debugging which obje

Re: how to propagate watermarks across multiple jobs

Hey Yidan, KafkaShuffle is initially motivated to support shuffle data materialization on Kafka, and started with a limited version supporting hash-partition only. Watermark is maintained and forwarded as part of shuffle data. So you are right, watermark storing/forwarding logic has nothing to do

Re: [DISCUSS] Deprecation and removal of the legacy SQL planner

big +1 from my side. Best, Jark On Thu, 4 Mar 2021 at 20:59, Leonard Xu wrote: > +1 for the roadmap. > > Thanks Timo for driving this. > > Best, > Leonard > > > 在 2021年3月4日，20:40，Timo Walther 写道： > > > > Last call for feedback on this topic. > > > > It seems everyone agrees to finally complete

Re: Stateful functions 2.2 and stop with savepoint

It doesn't change much ;) There is a known issue of stopping with savepoint and stateful functions not working [1]. The difference is that this one we will probably want to tackle sooner or later. Old streaming iterations are probably dead.. Piotrek [1] https://issues.apache.org/jira/browse/FLINK

Re: Stateful functions 2.2 and stop with savepoint

Hi all, My BAD!!! Sorry for apparent mess up in that moment. I will write a separate test for stream iterations. The stateful function part should be a separated issue. Best, Kezhu Wang On March 4, 2021 at 22:13:48, Piotr Nowojski (piotr.nowoj...@gmail.com) wrote: Hi Meissner, Can you cl

Watermark doesn't progress after job restore from savepoint

Hi All, I have a job where my source is kafka. Stream1 is partition the data on dynamic key, join the data with static rules(source kafka).I use KeyedCoProcessFunction to join the Steam1 with Stream2(source kafka). All works fine in a normal run. For changing the watermark generation interval I s

Re: Compile Error

Hi Chesnay, Thank you for your reply. Sorry for coming up late. I follow the maven quickstart. when I run the pom.xml in m2Maven clean way. Then it gives me this output in console: SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/Users/abdullah/.p2/pool/plu

Re: Defining GlobalJobParameters in Flink Unit Testing Harnesses

Thanks Chesnay! I tried giving that a shot but I still wasn't able to access the globalJobParameters from within the open function in my KeyedProcessFunction. You can see the implementation below which I believe should be correct: object CustomProcessFunctionTestHarness { fun forKeyedProcess

Re: Stateful functions 2.2 and stop with savepoint

Hi Meissner, Can you clarify, are you talking about stateful functions? [1] Or the stream iterations [2]? The first e-mail suggests stateful functions, but the ticket that Kezhu created is talking about the latter. Piotrek [1] https://flink.apache.org/news/2020/04/07/release-statefun-2.0.0.html

Re: Defining GlobalJobParameters in Flink Unit Testing Harnesses

The reason why your attempts have failed is that ProcessFunctionTestHarnesses.forKeyedProcessFunction automatically calls open(), thus any mutations on the harness happen too late. I'd suggest to take a look at the implementation of that method and essentially copy the code. You can then call

Re: Defining GlobalJobParameters in Flink Unit Testing Harnesses

Absolutely, I think it's gone through quite a few iterations, but this is the current state of it (defined in a @Before function as part of scaffolding out the tests): private lateinit var magicWindowHarness: KeyedOneInputStreamOperatorTestHarness @Before fun init() { magicWindowHarness = Pr

Convert BIGINT to TIMESTAMP in pyflink when using datastream api

Hello, We are using pyflink's datastream api v1.12.1 to consume from kafka and want to use one of the fields to act as the "rowtime" for windowing. We realize we need to convert BIGINT to TIMESTAMP before we use it as "rowtime". py4j.protocol.Py4JJavaError: An error occurred while calling o91.sel

Re: [DISCUSS] Deprecation and removal of the legacy SQL planner

+1 for the roadmap. Thanks Timo for driving this. Best, Leonard > 在 2021年3月4日，20:40，Timo Walther 写道： > > Last call for feedback on this topic. > > It seems everyone agrees to finally complete FLIP-32. Since FLIP-32 has been > accepted for a very long time, I think we don't need another votin

Re: apache-flink +spring boot - logs related to spring boot application start up not printed in file in flink 1.12

@Matthis tried this but did not work, normal logs (application logs) are coming. but the startup bean creation or server error log in case of build failure are not getting printed in file -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Defining GlobalJobParameters in Flink Unit Testing Harnesses

Could you show us how you create test harness? On 3/4/2021 5:13 AM, Rion Williams wrote: Hi all, Early today I had asked a few questions regarding the use of the many testing constructs available within Flink and believe that I have things in a good direction at present. I did run into a spec

Re: [DISCUSS] Deprecation and removal of the legacy SQL planner

Last call for feedback on this topic. It seems everyone agrees to finally complete FLIP-32. Since FLIP-32 has been accepted for a very long time, I think we don't need another voting thread for executing the last implementation step. Please let me know if you think differently. I will start

Re: How to emit after a merge?

Yes, implementing a UDF might be the most convenient option for some use cases. The accumulator of such a UDF could take the two timestamps and perform the two aggregations at once. The upsert-kafka connector can apply the updates to the Kafka log. If you enable log compaction in Kafka, Kafka

Re: Processing-time temporal join is not supported yet.

Sorry, I mean you can create a UDTF where you can cache data from your files and then enrich your stream with LATERAL TABLE grammar, BTW, you can reference FileSystemLookupFunction.java[1]. If we plan to support Lookup for filesystem, we should use this function too. Best, Leonard [1] https:/

Re: Processing-time temporal join is not supported yet.

Thx Leonard, by UDF you mean a custom table source on s3? Le jeu. 4 mars 2021 à 05:31, Leonard Xu a écrit : > Hi, Eric > > what will be the best workaround to enrich stream of data from a kafka > topics with statical data based on id? > > Currently you can put your statical data in Hive/JDBC/HBa

AW: State Schema Evolution within SQL API

Hi Qinghui Maybe this strategy with evolving an application without the need for state restore could help you [1] [1] https://docs.ververica.com/v2.3/user_guide/sql_development/sql_scripts.html#sql-script-changes Best, Jan Von: XU Qinghui Gesendet: Donnerstag, 4. März 2021 10:18 An: Dawid

AW: Flink upgrade causes operator to lose state

Maybe this strategy with evolving an application without the need for state restore could help you https://docs.ververica.com/v2.3/user_guide/sql_development/sql_scripts.html#sql-script-changes -Ursprüngliche Nachricht- Von: Chesnay Schepler Gesendet: Mittwoch, 3. März 2021 21:35 An: so

Re: How to emit after a merge?

Hi Timo, Thanks for the reply! > You could filter the deletions manually in DataStream API before writing them to Kafka. Yah I agree this helps the issue, though I will need to mix up SQL and DataStream API. > To simplify the query you could also investigate to implement your own aggregate func

Re: Timeout Exception When Producing/Consuming Messages to Hundreds of Topics

Hi Claude, Are you sure that the level of abstraction is fitting to your application? Maybe you could describe your use case a bit. Since you use one consumer, I'm assuming that all topics have the same schema. Then, for me, hundreds of similarly structured topics with X partitions sound counter-

Re: how to propagate watermarks across multiple jobs

And do you know when kafka consumer/producer will be re implemented according to the new source/sink api? I am thinking whether I should adjust the code for now, since I need to re adjust the code when it is reconstructed to the new source/sink api. yidan zhao 于2021年3月4日周四 下午4:44写道： > I uploaded

Re: State Schema Evolution within SQL API

Hello Dawid I'm interested in this discussion because I'm currently trying to upgrade flink from 1.9 to 1.12 for a bunch of sql jobs running in production. From what you said, this seems to be a risky operation which might corrupt the states. Is there any recommendation to mitigate the risk? It wo

Re: Flink + Hive + Compaction + Parquet?

Hi, I know Jingsong worked on Flink/Hive filesystem integration in the Table/SQL API. Maybe he can shed some light on your questions. Best, Dawid On 02/03/2021 21:03, Theo Diefenthal wrote: > Hi there, > > Currently, I have a Flink 1.11 job which writes parquet files via the > StreamingFileSink

Re: Job downgrade

Hi Alexey, Are you using the heap backend? If that's the case, then for whatever state was registered at the time of a savepoint, Flink will attempt to restore it to the heap backends. This essentially means that state "B" will be read as well, that would explain why Flink is trying to locate clas

Re: Scaling Higher than 10k Nodes

Maybe a stupid question Joey, but if the problem is in the resource managers, haven't you tried running standalone Flink clusters without any resource manager? Probably you would still hit the JobManager problems that Xintong mentioned, but those problems we can help addressing. Piotrek czw., 4 m

Re: State Schema Evolution within SQL API

Hi Jan, As of now Flink does not give any guarantees for Table/SQL API savepoint compatibility if you change the query or Flink version. Flink Table/SQL API uses an optimizer that can apply different optimizations or operations reordering based on the queried fields or computations that can result

Re: how to propagate watermarks across multiple jobs

I uploaded a picture to describe that. https://ftp.bmp.ovh/imgs/2021/03/2068f2e22045e696.png >

Re: timeWindow()s and queryable state

Hey Ron, I am pretty sure the queryable state will not do any pruning. It will keep the state for all windows seen so far. The allowedLateness applies to the window computation not the queryable state part. The `asQueryableState` will create a downstream operator that will keep updating a state wi

Re: Timeout Exception When Producing/Consuming Messages to Hundreds of Topics

Hi, Sorry, I don't know. I've heard that this kind of pattern is discouraged by Confluent. At least it used to be. Maybe someone else from the community will be able to help from his experience, however keep in mind that under the hood Flink is just simply using KafkaConsumer and KafkaProducer pr

Re: How to emit after a merge?

Hi Yik, if I understand you correctly you would like to avoid the deletions in your stream? You could filter the deletions manually in DataStream API before writing them to Kafka. Semantically the deletions are required to produce a correct result because the runtime is not aware of a key fo

Re: Scaling Higher than 10k Nodes

Hi Joey, Quick question: by *nodes*, do you mean Flink task manager processes, or physical/virtual machines (like ecs, yarn NM)? In our production, we run flink workloads on several Yarn/Kubernetes clusters, where each cluster typically has 2k~5k machines. Most Flink workloads are deployed in sin

Re: Producer Configuration

Hey Claude. Alexey is right about the page. The page from your screenshot shows only the entries passed via StreamExecutionEnvironment#getConfig#setGlobalJobParameters. Configuration for individual connectors or other operators is not displayed there. If you need help debugging your time out issu

Re: how to propagate watermarks across multiple jobs

One more question, If I only need watermark's logic, not keyedStream, why not provide methods such as writeDataStream and readDataStream. It uses the similar methods for kafka producer sink records and broadcast watermark to partitions and then kafka consumers read it and regenerate the watermark.