Re: Kubernetes HA Services - artifact for KubernetesHaServicesFactory

Yang, It worked finally... after adding kubernetes related config in configmap. On Thu, Jan 21, 2021 at 7:13 PM Yang Wang wrote: > You could set config option "kubernetes.namespace" to your flink-conf > ConfigMap. And then > KubernetesHAService will use it to create/watch the ConfigMap. Please n

Re: What is the best way to have a cache of an external database in Flink?

I provided an answer on stackoverflow, where I said the following: A few different mechanisms in Flink may be relevant to this use case, depending on your detailed requirements. *Broadcast State* Jaya Ananthram has already covered the idea

Re: A few questions about minibatch

Hello, Does anyone have any more information here? Thanks! On Wed, Jan 20, 2021 at 9:13 PM Rex Fenley wrote: > Hi, > > Our job was experiencing high write amplification on aggregates so we > decided to give mini-batch a go. There's a few things I've noticed that are > different from our previo

AW: What is the best way to have a cache of an external database in Flink?

But then you need a way to consume a database as a DataStream. I found this one https://github.com/ververica/flink-cdc-connectors. I want to implement a similar use case, but I don’t know how to parse the SourceRecord (which comes from the connector) into an PoJo for further processing. Best,

DataStream API: Best way for reading csv file

Hi , i'm looking for an comfortable way to read a CSV file with the DataStream API in Flink 1.11. Without using the Table/SQL-API before. This is my first approach: val typeInfo = TypeInformation.of(classOf[CovidEvent]).asInstanceOf[PojoTypeInfo[CovidEvent]] val csvInputFormat = new PojoCsv

Re: [Flink SQL] CompilerFactory cannot be cast error when executing statement

Thanks a lot Matthias! In the meantime I'm trying out something with the scala quickstart. On Fri, 22 Jan 2021 at 17:12, Matthias Pohl wrote: > Ok, to be fair, I just did some research on the error message and didn't > realize that you're working with binaries only. > > I tried to set it up on

Re: [Flink SQL] CompilerFactory cannot be cast error when executing statement

Ok, to be fair, I just did some research on the error message and didn't realize that you're working with binaries only. I tried to set it up on my machine to be able to reproduce your error. Unfortunately, I wasn't able to establish the connection between Flink and Postgres using your docker-comp

Re: What is the best way to have a cache of an external database in Flink?

Hi, Perhaps broadcast state is natural fit for this scenario. https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/state/broadcast_state.html Thanks, Selvaraj C On Fri, 22 Jan 2021 at 8:45 PM, Kumar Bolar, Harshith wrote: > Hi all, > > The external database consists of a se

Re: [Flink SQL] CompilerFactory cannot be cast error when executing statement

Hi Matthias! I went through that thread but as I'm just using the `apache/flink` docker image for testing I honestly couldn't figure out how I would do that since I don't have a pom file to edit. If it's possible to do it through the configuration I'd be glad if you could point me out in the right

Re: [Flink SQL] CompilerFactory cannot be cast error when executing statement

Hi Sebastián, have you tried changing the dependency scope to provided for flink-table-planner-blink as it is suggested in [1]? Best, Matthias [1] http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/Flink-1-10-exception-Unable-to-instantiate-java-compiler-td38221.html On Fri, Jan 22,

What is the best way to have a cache of an external database in Flink?

Hi all, The external database consists of a set of rules for each key, these rules should be applied on each stream element in the Flink job. Because it is very expensive to make a DB call for each element and retrieve the rules, I want to fetch the rules from the database at initialization and

[Flink SQL] CompilerFactory cannot be cast error when executing statement

Hi! I'm trying out Flink SQL with the attached docker-compose file. It starts up and then I create a table with the following statement: CREATE TABLE mytable_simple ( `customer_id` INT ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:postgresql://pgusr:pgpwd@postgres/pdgb', 'table-name' = 'm

Re: Flink 1.11 checkpoint compatibility issue

Hi Lu, if you are using data stream API make sure to set manual uids for each operator. Only then migrating of savepoints to other major versions of Flink is supported. [1] Best, Arvid [1] https://ci.apache.org/projects/flink/flink-docs-stable/ops/state/savepoints.html On Fri, Jan 22, 2021 at

Re: Unknown call expression: avg(amount) when use distinct() in Flink,Thanks~!

Hi, I'm assuming you are using Flink 1.12? The exception indicates that something is definitely going wrong with the translation from Table API to optimizer nodes. We refactored a lot of this code in this region. I investogate the issue and come back to you once I opended a ticket. Thanks f

RE: org.apache.flink.runtime.client.JobSubmissionException: Job has already been submitted

Hi Robert, I appreciate you having a look. I’ll have a closer look and see what I can find. Thanks! // ah From: Robert Metzger Sent: Friday, January 22, 2021 2:41 AM To: Hailu, Andreas [Engineering] Cc: user@flink.apache.org Subject: Re: org.apache.flink.runtime.client.JobSubmissionException:

Re: Flink 1.11 checkpoint compatibility issue

Hi Lu, thanks for reaching out to the community, Lu. Interesting observation. There's no change between 1.9.1 and 1.11 that could explain this behavior as far as I can tell. Have you had a chance to debug the code? Can you provide the code so that we could look into it more closely? Another thing:

File not generated using StreamingFileSink path 1.12.0

I’m trying to stream data to a file on an S3 compatible system (MINIO): DataStream resultStream = tEnv.toAppendStream(log_counts, Types.ROW(Types.STRING, Types.STRING, Types.LONG)); final StreamingFileSink sink = StreamingFileSink.forRowFormat( new Path("s3://argo-artifact

Unknown call expression: avg(amount) when use distinct() in Flink,Thanks~!

I'm testing https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/tableApi.html the part "Distinct aggregation on over window"(Ctrl+f and search the whole above string in above link please) test code distinctaggregation3.java https://paste.ubuntu.com/p/7HJ9W3hVVN/ POJO needed

Re: Re: Test failed in flink-end-to-end-tests/flink-end-to-end-tests-common-kafka

Hi Smile, Have you used a clean checkout? I second Robert's statement considering that the dependency you're talking about is already part of flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml. It also has the correct scope set both in master and release-1.12. Best, Matthias On Fr

Re: Handling validations/errors in the flink job

Hi Sagar, have you had a look at CoProcessFunction [1]? CoProcessFunction enables you to join two streams into one and also provide context to use SideOutput [2]. Best, Matthias [1] https://github.com/apache/flink/blob/c6997c97c575d334679915c328792b8a3067cfb5/flink-streaming-java/src/main/java/or

Re: JDBC connection pools

Hi Marco, have you had a look into the connector documentation ([1] for the regular connector or [2] for the SQL connector)? Maybe, discussions about connection pooling in [3] and [4] or the code snippets provided in the JavaDoc of JdbcInputFormat [5] help as well. Best, Matthias [1] https://ci.a

[ANNOUNCE] 1.12.1 may still produce corrupted checkpoints

Dear users, Unfortunately, the bug in the unaligned checkpoint that we fixed in 1.12.1 still occurs under certain circumstances, such that we recommend to not use unaligned checkpoints in production until 1.12.2. While the normal processing is not affected by this bug, a recovery with corrupted ch

Re: Question about setNestedFileEnumeration()

Hi Wayne, based on other mailing list discussion ([1]) you can assume that the combination of FileProcessingMode.PROCESS_CONTINUOUSLY and setting FileInputFormat.setNestedFileEnumeration to true should work as you expect it to work. Can you provide more context on your issue like log files? Which

Re:Re: Test failed in flink-end-to-end-tests/flink-end-to-end-tests-common-kafka

Yes, I've tried from both the root directory and the sub module. Neither or them works. And the error messages are the same. At 2021-01-21 23:22:12, "Robert Metzger" wrote: Since our CI system is able to build Flink, I believe it's a local issue. Are you sure that the build is failing when