Hi Hequn Cheng,
Finally I got the problem and find the way to define the correct WaterMark by
your advice, thank you very much.
The problem is that I set the watermark to the
waterMark = maxEventTime - lag
And the timeWindow is 10Seconds, But I generated the test records too quickly
so the 10000 records are all in the window duration(my bad).
So flink are waiting for new more numbers to close the window.
Another one question is why I set 'env.setParallelism(1)’ and run the code in
IDEA(mini Flink cluster) , but the getWatermark is called in 4 different
threads?
Which time is the getWaterMark function called? After the keyBy operation or
after the source operation?
> On 3 Mar 2018, at 15:28, Hequn Cheng <chenghe...@gmail.com> wrote:
>
> Hi sundy,
>
> The default parallelism is 4. It seems that your configuration does not take
> effect. You can try 'env.setParallelism(1)' to set the job parallelism.
> For watermark, you can refer to [1] and [2].
>
> PS: Others can not see your reply because you only reply to me. Try reply to
> all so that others can help you too :-)
>
> [1]
> https://ci.apache.org/projects/flink/flink-docs-master/dev/event_time.html#watermarks-in-parallel-streams
>
> <https://ci.apache.org/projects/flink/flink-docs-master/dev/event_time.html#watermarks-in-parallel-streams>
> [2]
> https://ci.apache.org/projects/flink/flink-docs-master/dev/event_timestamps_watermarks.html#timestamps-per-kafka-partition
>
> <https://ci.apache.org/projects/flink/flink-docs-master/dev/event_timestamps_watermarks.html#timestamps-per-kafka-partition>
>
> On Sat, Mar 3, 2018 at 3:03 PM, sundy <543950...@qq.com
> <mailto:543950...@qq.com>> wrote:
>
> Hi Hequn Cheng,
>
> Thanks for you advice, I think I found the problem. But I didn't know why.
>
> Firstly, let me introduce my operations, I run the code in IDEA with
> MinFlinkCluster, and then I set the env parallelism to 1, and the kafka
> topic ‘stream_test' has one partition. Then send 10000 records with current
> timestamp the kafka , format is ‘4281_783_1520059217832’ , the last field
> is the evenTime.
>
> I add this debug code to print the thread-id.
>
> <PastedGraphic-1.png>
>
>
> The result is it will print in 4 threads int each period duration(3
> seconds).Such as
>
> 50:water mark:0
> 52:water mark:0
> 51:water mark:1520056427871
> 49:water mark:0
>
> So this results to the watermark 0. But why it happened in 1 parallelism?
> Maybe it was caused by the keyBy operation? I am new to Flink, I hope to
> know how to set the watermark in the right way.
>
>
>
>
>
>> On 3 Mar 2018, at 13:52, Hequn Cheng <chenghe...@gmail.com
>> <mailto:chenghe...@gmail.com>> wrote:
>>
>> Hi sundy,
>>
>> 1. Some partition of your input kafka don't have data. Since window
>> watermark is the min value of all it's inputs, if there are no data from one
>> of it's inputs, window will never be triggered. You can set parallelism of
>> your job to 1 to avoid this problem(PS: Maybe this bug is fixed now, but
>> worth a try).
>> 2. Only one record in the input. In this case, window can not be triggered
>> either. You might think of it like the time has be stopped. To trigger the
>> widow, you should read more data with watermark bigger than the window end.
>>
>> Hope it helps you.
>> Best, Hequn
>>
>> 2018-03-03 13:06 GMT+08:00 sundy <543950...@qq.com
>> <mailto:543950...@qq.com>>:
>> Hi, thanks for your reply.
>>
>> I have searched it in stackoverflow, and there is someone who has the some
>> problem.
>>
>> https://stackoverflow.com/questions/40993753/flink-streaming-program-runs-correctly-with-processing-time-but-will-not-produc
>>
>> <https://stackoverflow.com/questions/40993753/flink-streaming-program-runs-correctly-with-processing-time-but-will-not-produc>
>>
>>
>> From your advice, I tried the code.
>>
>> env.getConfig().setAutoWatermarkInterval(3 * 1000);
>>
>> And it calls the getCurrentWaterMark function each 3 seconds, but still no
>> result come out.
>> From the outputs ('water mark1520049229163'), I could see that the add
>> method is called. But the no result from the sink function.
>>
>>
>>
>>
>>> On 3 Mar 2018, at 12:47, Xingcan Cui <xingc...@gmail.com
>>> <mailto:xingc...@gmail.com>> wrote:
>>>
>>> Hi,
>>>
>>> for periodically generated watermarks, you should use
>>> `ExecutionConfig.setAutoWatermarkInterval()` to set an interval.
>>>
>>> Hope that helps.
>>>
>>> Best,
>>> Xingcan
>>>
>>>> On 3 Mar 2018, at 12:08 PM, sundy <543950...@qq.com
>>>> <mailto:543950...@qq.com>> wrote:
>>>>
>>>>
>>>>
>>>> Hi, I got a problem in Flink and need your help.
>>>>
>>>> I tried to use TimeCharacteristic.EvenTime, but the sink function never be
>>>> executed.
>>>>
>>>> public class StreamingJob {
>>>> public static void main(String[] args) throws Exception {
>>>> // set up the streaming execution environment
>>>> final StreamExecutionEnvironment env =
>>>> StreamExecutionEnvironment.getExecutionEnvironment();
>>>> ObjectMapper jsonMapper = new ObjectMapper();
>>>>
>>>> Properties properties = new Properties();
>>>> // String brokers = "172.27.138.8:9092 <http://172.27.138.8:9092/>";
>>>> String brokers = "localhost:9092";
>>>> properties.setProperty("bootstrap.servers", brokers);
>>>> properties.setProperty("group.id <http://group.id/>", "test_fink");
>>>> String topic = "stream_test";
>>>>
>>>> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>>>> FlinkKafkaConsumer010<BitRate> myConsumer =
>>>> new FlinkKafkaConsumer010(topic, new
>>>> BitRate.BitRateDeserializtionSchema(), properties);
>>>>
>>>> DataStream<BitRate> stream = env.addSource(myConsumer)
>>>> .assignTimestampsAndWatermarks(new CustomWatermarkEmitter());
>>>> DataStream<BitRate>
>>>> reduceItems =
>>>> stream
>>>> .keyBy(a -> a.gameId)
>>>> .timeWindow(Time.seconds(10))
>>>> .reduce((a, b) -> a.add(b));
>>>>
>>>> reduceItems.addSink(new FlinkKafkaProducer010<>(brokers,
>>>> "stream_sink", (tuple) -> {
>>>> try {
>>>> tuple.end();
>>>> System.out.println(tuple.rate + "\t" + tuple.user);
>>>> return jsonMapper.writeValueAsBytes(tuple);
>>>> } catch (JsonProcessingException e) {
>>>> e.printStackTrace();
>>>> return "".getBytes();
>>>> }
>>>> }));
>>>>
>>>> env.execute("Flink Streaming Java API Skeleton");
>>>> }
>>>>
>>>> }
>>>>
>>>>
>>>> Here is the CustomWatermarkEmitter. I tried to increase the lag number,
>>>> but not worked.
>>>>
>>>> public class CustomWatermarkEmitter implements
>>>> AssignerWithPeriodicWatermarks<BitRate> {
>>>>
>>>> private long currentMax = 0;
>>>> private long lag = 3600 * 1000; //not worked ,even though the lag is
>>>> very big
>>>>
>>>> @Nullable
>>>> @Override
>>>> public Watermark getCurrentWatermark() {
>>>> long atLeastTime = currentMax - lag;
>>>> System.out.println("water mark" + atLeastTime);
>>>> return new Watermark(atLeastTime < 0 ? 0 : atLeastTime);
>>>> }
>>>>
>>>> @Override
>>>> public long extractTimestamp(BitRate bitRate, long l) {
>>>> currentMax = Long.max(bitRate.eventTime, currentMax);
>>>> return bitRate.eventTime;
>>>> }
>>>> }
>>>>
>>>>
>>>> Here is the entity BitRate, the logs are generated in time , sample log
>>>> `4281_783_1520047769115`
>>>>
>>>>
>>>> public BitRate(long eventTime, long gameId, long rate, long user) {
>>>> this.eventTime = eventTime;
>>>>
>>>> this.gameId = gameId;
>>>> this.rate = rate;
>>>> this.user = user;
>>>> this.startTs = System.currentTimeMillis();
>>>> this.type = 0;
>>>> }
>>>>
>>>> public void end() {
>>>> this.endTs = System.currentTimeMillis();
>>>> }
>>>>
>>>> public BitRate add(BitRate b) {
>>>> System.out.println("Add:" + b.rate);
>>>> this.rate += b.rate;
>>>> this.user += b.user;
>>>> return this;
>>>> }
>>>>
>>>
>>
>>
>
>
