I have issues concerning the table-api of Flink (1.13+). I have a POJO containing several fields, one of them being: List<String> my_list; I create my table using the following declaration ...

My Flink pipeline currently uses an Pojo that contains some Lists and Maps (of Strings), along the lines of public class MyPojo { private List<String> myList = new ArrayList<>(); pri...

I'm wanting to run a continuous stream processing job using Beam on a Flink runner within Kubernetes. I've been following this tutorial here (https://python.plainenglish.io/apache-beam-flink-cluste...

I am looking for the simplest possible example of an hello-world experience with Apache flink. Assume I have just installed flink on a clean box, what is the bare minimum I would need to do to 'mak...

I'm studying distributed systems and referring to this old question: stackoverflow link I really can't understand the difference between exactly-once, at-least-once and at-most-once guarantees, I ...

While using Apache Flink with the following code: DataStream<List<String>> result = source.window(Time.of(1, TimeUnit.SECONDS)).mapWindow(new WindowMapFunction<String, List<Strin...

I am using pyflink 1.17.1 and i am getting this error "RuntimeError: java.lang.UnsupportedOperationException: A serializer has already been registered for the state; re-registration is not all...

So, I'm trying to enable EXACTLY_ONCE semantic in my Flink Kafka streaming job along with checkpointing. However I am not getting it to work, so I tried downloading the test sample code from Githu...

I am trying to create a JUnit test for a Flink streaming job which writes data to a kafka topic and read data from the same kafka topic using FlinkKafkaProducer09 and FlinkKafkaConsumer09 respectiv...

This is with Flink 1.13.2 running in Amazon's Kinesis Data Analytics Flink environment. This application is running on Kafka topics. When the topics had smaller traffic volumes, this application ra...

I'm trying to connect Flink to a Kafka consumer I'm using Docker Compose to build 4 containers zookeeper, kafka, Flink JobManager and Flink TaskManager. For zookeeper and Kafka I'm using wurstmei...

I'm trying to create a source table using Apache Flink 1.11 where I can get access to nested properties in a JSON message. I can pluck values off root properties but I'm unsure how to access nested...

Context We are using Flink to run a number of streaming jobs that read from Kafka, perform some SQL transformation and write the output to Kafka. It runs on Kubernetes with two jobmanagers and many...

I would like to access the Flink Web interface when starting it locally from the IDE. I need this because I'd like to access the counters (accumulators) of Flink.

I am trying to stream data from kafka using flink. My code compiles without error but on running I get the following error: Error: A JNI error has occurred, please check your installation and try...

Does AWS Schema Registry support being used as an SQL Catalog within Flink SQLK applications? For instance, the documentation shows an example of using a Hive Catalog: CREATE CATALOG hive WITH ( 't...

I have a flink application which reads from kafka and sinks it to kafka. Application runs with no issues when i run it from Intellij IDEA, but gives ClassCastException when I submit the shadowJar t...

Currently, I have a running Flink Kubernetes session cluster (Flink version 1.13.2) and I can access the web UI by port-forward also, I can submit the WordCount jar example by this command ./bin/fl...

first of all I have read this post about the same issue and tried to follow the same solution that works for him (create a new quickstart with mvn and migrate the code there) and is not working eig...

Apache Beam supports multiple runner backends, including Apache Spark and Flink. I'm familiar with Spark/Flink and I'm trying to see the pros/cons of Beam for batch processing. Looking at the Beam...

I'm trying to pass our streaming pipeline to Table API and I almost did it except one field. I'm reading CSV data from Kafka topic then I'm doing some transformation and sending transformed data to...

This question covers how to sort an out-of-order stream using Flink SQL, but I would rather use the DataStream API. One solution is to do this with a ProcessFunction that uses a PriorityQueue to bu...

I have a simple Flink streaming app. It runs well in a cluster created by start-cluster.sh command. Now based on the Flink tutorial, I hope to deploy it in application mode natively in a Kubernetes...

I am trying to perform a map operation on a KeyedStream in Flink: stream.map(new JsonToMessageObjectMapper()) .keyBy("keyfield") .map(new MessageProcessorStateful()) The output of the JsonToOb...

I was trying to submit a Flink job to a cluster: ./bin/flink run -m <ip>:8081 examples/batch/WordCount.jar --input /opt/flink/README.txt but got the error Failed to deserialize JobGraph: ...