flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From 刘宇宝 <liuyu...@yingmi.cn>
Subject Re: how to hold a stream until another stream is drained?
Date Fri, 10 Apr 2020 14:21:05 GMT
Hi Fabian,

Thank you very much,  I almost get it working with InputSelectable:

       DataStream binlogStream = env.addSource(new FlinkKafkaConsumer(…));
       DataStream snapshotStream = env.createInput(JDBCInputFormat.buildJDBCInputFormat()….);
       DataStream tableStream = snapshotStream.connect(binlogstream);
       tableStream.transform(“Concat”, new TypeHint<….>(){},   new SequentialReadingStreamOperator<>());

The “SequentialReadingStreamOperator” is basically copied from https://github.com/apache/flink/blob/release-1.10.0/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestSequentialReadingStreamOperator.java

But if I enable checkpoining with “streamEnv.enableCheckpointing(10000);”,   Flink throws
exception below,   any idea to resolve that? 

Caused by: java.lang.UnsupportedOperationException: Checkpointing is currently not supported
for operators that implement InputSelectable:example.SequentialReadingStreamOperator
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.preValidate(StreamingJobGraphGenerator.java:219)
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:149)
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:104)
	at org.apache.flink.streaming.api.graph.StreamGraph.getJobGraph(StreamGraph.java:777)
	at org.apache.flink.streaming.api.graph.StreamGraphTranslator.translateToJobGraph(StreamGraphTranslator.java:52)
	at org.apache.flink.client.FlinkPipelineTranslationUtil.getJobGraph(FlinkPipelineTranslationUtil.java:43)
	at org.apache.flink.client.deployment.executors.ExecutorUtils.getJobGraph(ExecutorUtils.java:51)
	at org.apache.flink.client.deployment.executors.AbstractSessionClusterExecutor.execute(AbstractSessionClusterExecutor.java:57)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1733)
	at org.apache.flink.streaming.api.environment.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:94)
	at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:63)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1620)
	at example.Main.main(Main.java:72)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:564)
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321)
	... 8 more

Regards,
Yubao Liu

------------------------------------------------------------------------------------------------------------------------
From: Fabian Hueske <fhueske@gmail.com>
Date: Tuesday, April 7, 2020 at 4:45 AM
To: 刘宇宝 <liuyubao@yingmi.cn>
Cc: user <user@flink.apache.org>
Subject: Re: how to hold a stream until another stream is drained?

Hi,

With Flink streaming operators 

However, these parts are currently being reworked to enable a better integration of batch
and streaming use cases (or hybrid use cases such as yours).
A while back, we wrote a blog post about these plans [1]:

> "Unified Stream Operators: Blink extends the Flink streaming runtime operator model to
support selectively reading from different inputs, while keeping the push model for very low
latency. This control over the inputs helps to now support algorithms like hybrid hash-joins
on the same operator and threading model as continuous symmetric joins through RocksDB. These
operators also form the basis for future features like https://cwiki.apache.org/confluence/display/FLINK/FLIP-17+Side+Inputs+for+DataStream+API."

I'm not familiar with the internal details here, but I found the InputSelectable [2] interface
that looks like it would do what you are looking for.
Note that this interface is not used on the higher-level DataStream API level, but rather
on the lower StreamOperator level.

Best, Fabian

[1] https://flink.apache.org/news/2019/02/13/unified-batch-streaming-blink.html
[2] https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InputSelectable.java




W

Am Mo., 6. Apr. 2020 um 12:49 Uhr schrieb 刘宇宝 <mailto:liuyubao@yingmi.cn>:
I’m using JDBCInputFormat to read snapshot of a MySQL table  and FlinkKafkaConsumer to
read binlog which is written to Kafka by Debezium.
 
       DataStream binlogStream = env.addSource(new FlinkKafkaConsumer(…));
       DataStream snapshotStream = env.createInput(JDBCInputFormat.buildJDBCInputFormat()….);
        
       // map() is to convert two streams into same type:  (action,  fields…),  where
action is “insert”, “update”, “delete”.  The action for “snapshotStream”
is always “insert”.
       DataStream tableStream = binlogStream.map(…).union(snapshotStream.map(…));
        
       tableStream.print();
       env.execute(“example”);
 
1. To make sure “tableStream” doesn’t miss any row,  the “binlogStream” must connect
to  Kafka first so that binlog starts before the table snapshot,  I can roughly achieve
this by “myKafkaConsumer.setStartFromTimestamp(System.currentTimeMillis() – 600*1000)”.
2. To make sure changes from “binlogStream” always overwrite upon “snapshotStream”,  
I need a way to hold “binlogStream”  until “snapshotStream” is drained,  so that
changes from “binlogStream” are all behind changes from “snapshotStream”.  How can
I achieve this ?
 
I’m considering a wrapper SourceFunction to combine FlinkKafkaConsumer and JDBCInputFormat, 
but they are different on parallelism  and checkpointing,  I’m not sure how to get the
wrapper right and even whether it’s right direction.
 
Any suggestion will be very appreciated!
 

Mime
View raw message