Hello I have a dummy question I have this Job which sink to Apache Flink #troubleshooting

Hello I have a dummy question, I have this Job whi...

Adrian Chang

02/21/2023, 5:31 PM

Hello I have a dummy question, I have this Job which sink to Kafka and Redis. I am using Table API for Kafka connector because I need to use

format = "avro-confluent"

Copy code

// Sink Motion Detection to Kafka
Table motionDetectionTable = tableEnv.fromDataStream(motionDetectionStream);
motionDetectionTable.executeInsert(Util.MOTION_DETECTION_SINK_TABLE_NAME);

// Sink stream.motion messages to Redis
motionStream.addSink(new RedisSink<>(Util.getRedisConfig(), new RedisMotionMapper()));

env.execute("Motion");

When the Job is deployed, two Task Managers starts ( I have parallelism set to 1 ) One execution graph is completely contained into the other, causing calculation being duplicated and producing duplicates to Kafka. I suspect it's because I am calling

executeInsert

and

env.execute

, but if I don't call them both the Job doesn't produce to both sink connectors. Am I missing something ? Thanks

Adrian Chang

02/22/2023, 4:14 PM

it's not producing duplicates to Kafka I guess is not computing twice so I think this is the normal behaviour based on

To combine both execution behaviors, every call to
StreamTableEnvironment.toDataStream
or
StreamTableEnvironment.toChangelogStream
will materialize (i.e. compile) the Table API sub-pipeline and insert it into the DataStream API pipeline builder. This means that
StreamExecutionEnvironment.execute()
or
DataStream.executeAndCollect
must be called afterwards. An execution in Table API will not trigger these “external parts”.

Open in Slack

Previous Next