Questions tagged [flink-streaming]

Apache Flink is an open source platform for scalable batch and stream data processing. Flink supports batch and streaming analytics, in one system. Analytical programs can be written in concise and elegant APIs in Java and Scala.

Filter by
Sorted by
Tagged with
1 vote
0 answers
13 views

TaskManagers not scaling in flink Session Deployment using HPA

I've set up a Flink cluster operating in session mode, with Kubernetes underneath. I'm using FlinkK8Operator with helm charts to launch my jobs. I've configured a simple HPA with the deployment, with ...
Riyan Mohammed's user avatar
0 votes
0 answers
15 views

Implementing TypeInformation in Flink for a class with an interface field

I have a fairly nested data type that comes into flink over a Kafka topic. The JSON is deserialized into a Java class hierarchy that uses interfaces (together with JsonSubTypes, see https://www....
dashambles's user avatar
0 votes
0 answers
23 views

Flink application running in kubernetes failing with ProducerFencedException

I have a flink application where I am setting execution.checkpointing.mode='EXACTLY_ONCE' and then I have a KafkaSink with the following setup KafkaSink<KafkaSinkEvent> sink = ...
Parag Panda's user avatar
0 votes
0 answers
9 views

HybridSource with AvroInputFormat in Flink

How do I use a Hybridsource with the AvroInputFormat? HybridSource requires a Source but AvroInputFormat extends FileInputFormat which is not accepted by FileSource. It looks like those are two ...
Clemens Valiente's user avatar
0 votes
0 answers
30 views

Flink job ingests kafka data very slowly, while no resources or backpressure to slow it down. What could be the reason?

We have 5 Flink streaming jobs that ingest and process data from 5 Kafka queues respectively. The jobs are running on the same cluster, with 1 job manager, 6 task managers and 42 parallelism. We've ...
wd_zhong's user avatar
0 votes
0 answers
25 views

Flink OOM error processing of large sized records

We have a Flink application which has Kafka topics as source and S3 bucket as a FileSinknk. The aim of the pipeline is to read data from Kafka Topics and store it in parquet files for analytics. The ...
davyjones's user avatar
  • 185
1 vote
0 answers
48 views

Flink + rocksdb taking a very long time to restore a large s3 checkpoint

I am running a low-parallellism (4 slots) job whose checkpoints can get very large. In the example I'm going to present the checkpoint was 142 GB, saved in S3 and it took 40 minutes to restore from ...
Raúl García's user avatar
0 votes
0 answers
19 views

Modifying State in Flink's Aggregate Function within a Windowed Operation

I'm working on a Flink application where I'm using an aggregate function over a window. I've been able to successfully read the state of its outputs using the provided APIs. However, in my specific ...
Jakub Berezowski's user avatar
0 votes
1 answer
48 views

Apache Flink same source stream writing to different sinks

I have a DataStream<Event> dataStream and I'd like to have 2 ProcessFunction to process the same stream and write to 2 sinks respectively. I might have 2 options - Use 2 process functions on ...
Muzzy's user avatar
  • 21
0 votes
1 answer
21 views

Is it safe to modify inputs in flink aggregation?

Is it safe to modify input objects in flink aggregation function? Sample code: class Acc { int sum = 0; int max = 0; } public class MyAggFunc implements AggregateFunction<Integer, Acc, Acc&...
user50874's user avatar
  • 311
0 votes
1 answer
26 views

How can I send my data to Kafka where data specific to a key goes to same partition only using KafkaSink in a Flink Streaming Job?

I have a requirement where I wish to send my data to same partition only if the data has same key. For example: {"field1": 33, "field2": 44, "field3": 55, "...
Alok Singh's user avatar
0 votes
0 answers
20 views

Flink KDA truststore location not working

I am using AWS(Kinesis Data analytics) to deploy flink service to consume data using KafkaSource, while passing the ssl.truststore.location, receiving below "Caused by: org.apache.kafka.common....
user22555227's user avatar
0 votes
0 answers
17 views

How do I make my logs show up in yarn from my flink job?

I have a flink job that uses slf4j for logging but when I run the job, the logs do not show up in yarn. Do I need a log4j.properties or is there something else I have to configure? I am just doing ...
Scott's user avatar
  • 121
-1 votes
1 answer
28 views

getting java.lang.NoClassDefFoundError: org/apache/flink/shaded/guava30/com/google/common/collect/ImmutableList

I am using flink 1.17.1 and java version 11 . I am getting below error Caused by: java.lang.ClassNotFoundException: org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList Need ...
samanth mahankali's user avatar
0 votes
0 answers
38 views

Potential Causes for InvalidPidMappingException in Transactional Flink Job

Currently, I have an existing Flink job that handles aggregating data from various sources, consolidating them into a common schema, and then perform various types of enrichment operations before ...
Rion Williams's user avatar
0 votes
1 answer
34 views

Can the results of datastream be output to List or Map?

I hope to read the ID field in the database table through a stream, store the field in the list, and then create a new stream based on this ID to filter the corresponding data and synchronize it to ...
han liu's user avatar
0 votes
0 answers
31 views

High velocity streaming data enrichment with low velocity/slow changing data

My system consists of High velocity telemetry data generated by IoT devices. Relatively static/slow changing reference/lookup data - Alarm Rules Each IoT device has 0 to 1 Alarm Rules. An Alarm Rule ...
OverflowStack's user avatar
0 votes
1 answer
35 views

How does one operator pass on event stream to another operator?

Operator one can pass on event data to operator two, in a pipeline, through GoLang channel, if implemented in GoLang. This helps avoid separation of concerns, amidst runtime failure. Does flink ...
overexchange's user avatar
  • 15.9k
0 votes
0 answers
41 views

No state latency improvement using local SSD in Flink

I am running a Flink 1.14.2 application that processes hundreds of thousands of events per minute. This data is partitioned and I have multiple state data structures to read/write. In case it is of ...
Raúl García's user avatar
0 votes
1 answer
21 views

Using keyed stream keys for preloading

I want to do preloading of some data from remote data source depending on the key available for the current operator. Keyed process function has ctx.getCurrentKey but we cannot use it in open method. ...
yokus's user avatar
  • 144
0 votes
1 answer
21 views

Why isn't Flink distributing my work evenly and how do I fix it?

I have a flink job that reads from a data source that has 5 partitions. I have my parallelism in the config file set to 100 for each task manager. In my input operation, it only uses 5/100 task ...
Scott's user avatar
  • 121
0 votes
2 answers
32 views

How to refresh the redis cache in Apache flink?

I have a requirement to read the data from redis cache in flink, but as per the requirements the cache data will be refreshed on an average every two hours. I was going through the documentation and ...
user13906258's user avatar
-1 votes
1 answer
37 views

Flink deduplication - How does out.collect() work with unbounded stream?

Below is the code: public class VerifyDuplicate { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment....
overexchange's user avatar
  • 15.9k
0 votes
1 answer
36 views

Facing SequenceNumber is treated as a generic type exception with flink-connector-kinesis-4.1.0 jar

I am trying to use flink-connector-kinesis-4.1.0 jar[https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kinesis/4.1.0-1.17] with flink 1.17 for integrating kinesis stream connector ...
priyadhingra19's user avatar
0 votes
2 answers
46 views

Flink flatMap() - NullPointerException

Below is the code: public class VerifyDuplicate { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment....
overexchange's user avatar
  • 15.9k
0 votes
1 answer
20 views

Not seeing Kinesis connector package in Flink 1.17

We were using Flink kinesis connector library of flink(1.16) package in our Flink job. Now we are planning to move to Flink 1.17 but I am seeing flink-kinesis-connector package got removed from Flink-...
priyadhingra19's user avatar
0 votes
0 answers
16 views

Rescaling simple join application in Flink

Good afternoon, I am currently working on the scalability of stateful operators in stream processing systems. I am particularly interested in the case of the 'join' operator, for which I have ...
ark's user avatar
  • 1
0 votes
0 answers
17 views

How to Implement a Custom Health Check for Flink Task Managers to Automatically Restart on Connectivity Issues?

I recently encountered a problem with our Flink Task Managers running in a Kubernetes environment managed by the Flink Kubernetes Operator. These Task Managers intermittently lost connectivity to an ...
Techie97's user avatar
0 votes
0 answers
17 views

Table result to String

I am trying Dynamic rule with Table API, In my system I have SQL rule coming from Kafka. So I have used a broadcast mechanism to implement the same. But I am stuck on how can I give results back to ...
Keyur Makwana's user avatar
1 vote
1 answer
47 views

Flink - How many watermarks get injected in this bounded stream?

In the below code: StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); SingleOutputStreamOperator<Tuple2<String, Integer>> dataStream = env....
overexchange's user avatar
  • 15.9k
0 votes
1 answer
34 views

Flink - Why the bounded stream print out of order sequence?

In the below code: StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<Tuple2<String, Integer>> dataStream = env.fromElements( ...
overexchange's user avatar
  • 15.9k
0 votes
1 answer
35 views

Flink - When not to inject watermarks? for windowing

I learnt that, Windowing of any stream data uses water marks, as boundaries In the below code from answer, there is no water mark injected in the generated stream(fromElements()) case class Record( ...
overexchange's user avatar
  • 15.9k
0 votes
0 answers
9 views

In flink, converting from datastream to table performance implications

does convert from datastream to table and back from table to datastream impact performance of an app in a negative way? Tried looking for the flink datastream to table conversion on flink ...
Newbie's user avatar
  • 1
0 votes
1 answer
36 views

What should be the value of state.checkpoints.num-retained with Incremental checkpointing enabled

As per Flink documentation by default value of state.checkpoints.num-retained is 1 which means Flink only retails one checkpoint in memory. We have enabled Incremental checkpointing in our job. I ...
priyadhingra19's user avatar
0 votes
0 answers
76 views

How to include multiple filters in filter.includes parameter for my flink metric reporter in flink-conf.yaml?

I want to understand the working of a generic flink metrics reporter and I am trying out the filter.includes parameter. How do I specify multiple filters for this option in flink-conf.yaml? The docs ...
NR4276's user avatar
  • 83
0 votes
1 answer
21 views

Does flink always returns the same partitioned state objects for the same key value?

If we fill a state object inside a KeyedProcessFunction of a KeyedStream#process such as new KeyedProcessFunction<String, Rule, Rule>() { private MapState<String, ...
yokus's user avatar
  • 144
0 votes
0 answers
13 views

Is there a way to achieve partitioning while using Flink Broadcast pattern?

This example provides a use case where an operator needs whole elements of a first datastream while processing one element of the other (second) datastream. Is broadcasting pattern only way to achieve ...
yokus's user avatar
  • 144
0 votes
1 answer
22 views

How to filter a a flink DataStream depending on another?

I am trying to learn Apache Flink and use case mentioned below seems simple enough that flink not supporting it? makes me start to think I understand something fundamentally wrong. Use case is this ...
yokus's user avatar
  • 144
0 votes
0 answers
31 views

All Task managers crushes without any reason

I have 4 task manager working on 4 servers. They all crush at the same time without any useful error logs. Only log I can see is some disconnection from Kafka for both consumer and producers. Any idea ...
Kenank's user avatar
  • 321
0 votes
0 answers
41 views

Flink Serialization of java.util.Set

I'm having the same issue as described by Flink serialization of java.util.List and java.util.Map, but for java.util.Set. I have a POJO class that has List fields and Set fields. I've been able to ...
khan7's user avatar
  • 75
0 votes
0 answers
23 views

How to use parallelism in apache flink

I use zfrools and flink 1.17.1 to filter event, input/output are read/write file I keyBy(customerid) and in rule count events which same userid. If count >= 5 write events to file Result I set ...
thao nguyen's user avatar
6 votes
0 answers
70 views

RuntimeError: java.lang.UnsupportedOperationException: A serializer has already been registered for the state; re-registration is not allowed

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 ...
Imran Munir's user avatar
0 votes
0 answers
77 views

Problems with sink on flink

[enter image description here][1] I could do this to try first, this works well this just print the data on another topic I trie to follow the same logic but don get the result package org.example; ...
edu90s's user avatar
  • 3
0 votes
0 answers
32 views

Apache Flink throws exception when submitting a job through Jenkins and Spinnaker

We are using Apache Flink 1.16.1 configured as a standalone Kubernetes pod ,for one of our application to read from confluent Kafka topics to do event correlation. We are using the flink's Table API ...
JDP's user avatar
  • 35
0 votes
0 answers
9 views

Flink cep MATCH_RECOGNIZE not processing all events

I am using MATCH_RECOGNIZE for pattern matching in events, I am trying to find average usage less than 60 withing 10 seconds interval, the issue here is that in result i am only seeing only 2 result (...
Krutik Gadhiya's user avatar
0 votes
1 answer
45 views

Flink statebackend - how does parallelism work with RocksDB?

In the below code: package spendreport; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache....
overexchange's user avatar
  • 15.9k
0 votes
1 answer
34 views

Flink - Can statebackend allow purging old data in RocksDB table?

In the below code: class EnrichmentStream { private val checkpointsDir = "file://${System.getProperty("user.dir")}/checkpoints/" private val rocksDBStateDir = "file://...
overexchange's user avatar
  • 15.9k
0 votes
1 answer
15 views

flink table api cross table reference

We have two applications app 1 creates two tables ie sourceTable and targetTable1. It does select * from sourceTable to targetTable1. This are both created on kafka topics App2 creates a sourceTable2 ...
Kapil More's user avatar
0 votes
1 answer
34 views

Does statebackend need checkpoint?

Below is the code from here that maintains the checkpoint: class EnrichmentStream { private val checkpointsDir = "file://${System.getProperty("user.dir")}/checkpoints/" ...
overexchange's user avatar
  • 15.9k
0 votes
0 answers
26 views

Old Flink checkpoints not always correctly deleted

Summary We are using Flink 1.15.1 and have long-running stateful Flink jobs ingesting data from Kafka topics. They are configured to write checkpoints, with a RocksDB backend on S3. We have noticed ...
Colin Smetz's user avatar

1
2 3 4 5
64