flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-4174) Enhance Window Evictor
Date Thu, 03 Nov 2016 16:46:59 GMT

    [ https://issues.apache.org/jira/browse/FLINK-4174?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15633419#comment-15633419
] 

ASF GitHub Bot commented on FLINK-4174:
---------------------------------------

Github user aljoscha commented on a diff in the pull request:

    https://github.com/apache/flink/pull/2736#discussion_r86388455
  
    --- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorTest.java
---
    @@ -58,6 +61,458 @@
     
     	// For counting if close() is called the correct number of times on the SumReducer
     
    +	/**
    +	 * Tests CountEvictor evictAfter behavior
    +	 * @throws Exception
    +     */
    +	@Test
    +	public void testCountEvictorEvictAfter() throws Exception {
    +		AtomicInteger closeCalled = new AtomicInteger(0);
    +		final int WINDOW_SIZE = 4;
    +		final int TRIGGER_COUNT = 2;
    +		final boolean EVICT_AFTER = true;
    +
    +		TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String,
Integer>");
    +
    +		@SuppressWarnings({"unchecked", "rawtypes"})
    +		TypeSerializer<StreamRecord<Tuple2<String, Integer>>> streamRecordSerializer
=
    +			(TypeSerializer<StreamRecord<Tuple2<String, Integer>>>) new StreamElementSerializer(inputType.createSerializer(new
ExecutionConfig()));
    +
    +		ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc
=
    +			new ListStateDescriptor<>("window-contents", streamRecordSerializer);
    +
    +
    +		EvictingWindowOperator<String, Tuple2<String, Integer>, Tuple2<String,
Integer>, GlobalWindow> operator = new EvictingWindowOperator<>(
    +			GlobalWindows.create(),
    +			new GlobalWindow.Serializer(),
    +			new TupleKeySelector(),
    +			BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
    +			stateDesc,
    +			new InternalIterableWindowFunction<>(new RichSumReducer<GlobalWindow>(closeCalled)),
    +			CountTrigger.of(TRIGGER_COUNT),
    +			CountEvictor.of(WINDOW_SIZE,EVICT_AFTER),
    +			0);
    +
    +
    +		OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String,
Integer>> testHarness =
    +			new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(),
BasicTypeInfo.STRING_TYPE_INFO);
    +
    +
    +		long initialTime = 0L;
    +		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
    +
    +		testHarness.open();
    +
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1),
initialTime + 3000));
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1),
initialTime + 3999));
    +
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1),
initialTime + 20));
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1),
initialTime));
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1),
initialTime + 999));
    +
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1),
initialTime + 1998));
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1),
initialTime + 1999));
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1),
initialTime + 1000));
    +
    +
    +
    +		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 2), Long.MAX_VALUE));
    +		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 4), Long.MAX_VALUE));
    +		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 2), Long.MAX_VALUE));
    +
    +		TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput,
testHarness.getOutput(), new ResultSortComparator());
    +
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1),
initialTime + 10999));
    +		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1),
initialTime + 1000));
    +
    +		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 4), Long.MAX_VALUE));
    +		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 6), Long.MAX_VALUE));
    +
    +		TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput,
testHarness.getOutput(), new ResultSortComparator());
    --- End diff --
    
    As it is now, it doesn't seem this ever evicts elements because eviction happens after
triggering. If you add these lines that would exercise the eviction:
    
    ```
    testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), initialTime
+ 1000));
    testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), initialTime
+ 1000));
    expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 6), Long.MAX_VALUE));
    
    TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(),
new ResultSortComparator());
    ```
    
    It checks whether the previous firing with 6 elements truncated back to 4.


> Enhance Window Evictor
> ----------------------
>
>                 Key: FLINK-4174
>                 URL: https://issues.apache.org/jira/browse/FLINK-4174
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Streaming
>            Reporter: vishnu viswanath
>            Assignee: vishnu viswanath
>
> Enhance the current functionality of Evictor as per this [design document|https://docs.google.com/document/d/1Rr7xzlItYqvFXLyyy-Yv0vvw8f29QYAjm5i9E4A_JlU/edit].
> This includes:
> - Allow eviction of elements from the window in any order (not only from the beginning).
To do this Evictor must go through the list of elements and remove the elements that have
to be evicted instead of the current approach of : returning the count of elements to be removed
from beginning.
> - Allow eviction to be done before/after applying the window function.
> FLIP page for this enhancement : [FLIP-4|https://cwiki.apache.org/confluence/display/FLINK/FLIP-4+%3A+Enhance+Window+Evictor]



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message