You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2015/02/19 17:41:23 UTC
[2/6] flink git commit: [streaming] Add MultiEvictionPolicy: It wraps
around multiple eviction policies and allows to use them at the same time. It
supports both,
active and not active policies at the same time. Additionally it introduces
different evict
[streaming] Add MultiEvictionPolicy: It wraps around multiple eviction policies and allows to use them at the same time. It supports both, active and not active policies at the same time. Additionally it introduces different eviction strategies the user can choose from.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/09490c4b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/09490c4b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/09490c4b
Branch: refs/heads/master
Commit: 09490c4b95f04079b15c38c4b576f5beb861c038
Parents: 64f7bf7
Author: Jonas Traub (powibol) <jo...@s-traub.com>
Authored: Thu Feb 19 14:51:23 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Thu Feb 19 17:03:00 2015 +0100
----------------------------------------------------------------------
.../windowing/policy/MultiEvictionPolicy.java | 172 +++++++++++++++++
.../policy/MultiEvictionPolicyTest.java | 186 +++++++++++++++++++
2 files changed, 358 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/09490c4b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicy.java
new file mode 100644
index 0000000..d353dc8
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicy.java
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.policy;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * This policy provides the ability to use multiple eviction policies at the
+ * same time. It allows to use both, active and not active evictions.
+ *
+ * @param <DATA>
+ * The type of data-items handled by the policies
+ */
+public class MultiEvictionPolicy<DATA> implements ActiveEvictionPolicy<DATA> {
+
+ /**
+ * Default version id.
+ */
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This enum provides the different options for the eviction strategy.
+ *
+ * You can choose from the following:
+ * <ul>
+ * <li>MIN: The number of elements to evict will be the smallest one which
+ * is greater than 0 and was returned by any of the given policies. If all
+ * policies return 0, the result is 0.</li>
+ * <li>MAX: The number of elements to evict will be the greatest one which
+ * was returned by any of the given policies.</li>
+ * <li>SUM: The number of elements to evict will be the sum of all values
+ * returned by the nested eviction policies.</li>
+ * <li>PRIORITY: Depending on the order in which the policies have been
+ * passed to the constructor, the first return value greater than 0 will be
+ * the the number of elements to evict. If all policies return 0, the result
+ * is 0.</li>
+ * </ul>
+ */
+ public enum EvictionStrategy {
+ MIN, MAX, SUM, PRIORITY;
+ }
+
+ private List<EvictionPolicy<DATA>> allEvictionPolicies;
+ private List<ActiveEvictionPolicy<DATA>> activeEvictionPolicies;
+ private EvictionStrategy selectedStrategy;
+
+ /**
+ * This policy provides the ability to use multiple eviction policies at the
+ * same time. It allows to use both, active and not active evictions.
+ *
+ * When using this constructor the MAX strategy is used by default. You can
+ * select other strategies using
+ * {@link MultiEvictionPolicy#MultiEvictionPolicy(EvictionStrategy, EvictionPolicy...)}
+ * .
+ *
+ * @param evictionPolicies
+ * Any active or not active eviction policies. Both types can be
+ * used at the same time.
+ */
+ public MultiEvictionPolicy(EvictionPolicy<DATA>... evictionPolicies) {
+ this(EvictionStrategy.MAX, evictionPolicies);
+ }
+
+ /**
+ * This policy provides the ability to use multiple eviction policies at the
+ * same time. It allows to use both, active and not active evictions.
+ *
+ * @param strategy
+ * the strategy to be used. See {@link EvictionStrategy} for a
+ * list of possible options.
+ * @param evictionPolicies
+ * Any active or not active eviction policies. Both types can be
+ * used at the same time.
+ */
+ public MultiEvictionPolicy(EvictionStrategy strategy, EvictionPolicy<DATA>... evictionPolicies) {
+ // initialize lists of policies
+ this.allEvictionPolicies = new LinkedList<EvictionPolicy<DATA>>();
+ this.activeEvictionPolicies = new LinkedList<ActiveEvictionPolicy<DATA>>();
+
+ // iterate over policies and add them to the lists
+ for (EvictionPolicy<DATA> ep : evictionPolicies) {
+ this.allEvictionPolicies.add(ep);
+ if (ep instanceof ActiveEvictionPolicy) {
+ this.activeEvictionPolicies.add((ActiveEvictionPolicy<DATA>) ep);
+ }
+ }
+
+ // Remember eviction strategy
+ this.selectedStrategy = strategy;
+ }
+
+ @Override
+ public int notifyEviction(DATA datapoint, boolean triggered, int bufferSize) {
+ LinkedList<Integer> results = new LinkedList<Integer>();
+ for (EvictionPolicy<DATA> policy : allEvictionPolicies) {
+ results.add(policy.notifyEviction(datapoint, triggered, bufferSize));
+ }
+ return getNumToEvict(results);
+ }
+
+ @Override
+ public int notifyEvictionWithFakeElement(Object datapoint, int bufferSize) {
+ LinkedList<Integer> results = new LinkedList<Integer>();
+ for (ActiveEvictionPolicy<DATA> policy : activeEvictionPolicies) {
+ results.add(policy.notifyEvictionWithFakeElement(datapoint, bufferSize));
+ }
+ return getNumToEvict(results);
+ }
+
+ private int getNumToEvict(LinkedList<Integer> items) {
+ int result;
+ switch (selectedStrategy) {
+
+ case MIN:
+ result = Integer.MAX_VALUE;
+ for (Integer item : items) {
+ if (result > item) {
+ result = item;
+ }
+ }
+ return result;
+
+ case MAX:
+ result = 0;
+ for (Integer item : items) {
+ if (result < item) {
+ result = item;
+ }
+ }
+ return result;
+
+ case SUM:
+ result = 0;
+ for (Integer item : items) {
+ System.out.print(item+":");
+ result += item;
+ }
+ System.out.println(result);
+ return result;
+
+ case PRIORITY:
+ for (Integer item : items) {
+ if (item > 0) {
+ return item;
+ }
+ }
+ return 0;
+
+ default:
+ // The following line should never be reached. Just for the
+ // compiler.
+ return 0;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/09490c4b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicyTest.java
new file mode 100644
index 0000000..3214aa7
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicyTest.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.policy;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.junit.Test;
+
+public class MultiEvictionPolicyTest {
+
+ private final List<Integer> tuples = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+
+ private final CloneableEvictionPolicy<Integer> evictionPolicy1 = new CountEvictionPolicy<Integer>(
+ 7, 3);
+ private final CloneableEvictionPolicy<Integer> evictionPolicy2 = new CountEvictionPolicy<Integer>(
+ 3, 1);
+ private final CloneableEvictionPolicy<Integer> evictionPolicy3 = new CountEvictionPolicy<Integer>(
+ 5, 2);
+
+ private final CloneableEvictionPolicy<Integer> activeEvictionPolicy1 = new ActiveCloneableEvictionPolicyWrapper<Integer>(
+ evictionPolicy1);
+ private final CloneableEvictionPolicy<Integer> activeEvictionPolicy2 = new ActiveCloneableEvictionPolicyWrapper<Integer>(
+ evictionPolicy2);
+ private final CloneableEvictionPolicy<Integer> activeEvictionPolicy3 = new ActiveCloneableEvictionPolicyWrapper<Integer>(
+ evictionPolicy3);
+
+ // From policies specified above the expected output is:
+ // 1.: 0000000300
+ // 2.: 0001111111
+ // 3.: 0000020202
+ private final Integer[] maxResult = { 0, 0, 0, 1, 1, 2, 1, 3, 1, 2 };
+ private final Integer[] minResult = { 0, 0, 0, 0, 0, 0, 0, 1, 0, 0 };
+ private final Integer[] sumResult = { 0, 0, 0, 1, 1, 3, 1, 6, 1, 3 };
+ private final Integer[] priorityResult = { 0, 0, 0, 1, 1, 1, 1, 3, 1, 1 };
+
+ /*
+ * Test cases for not active policies
+ */
+
+ @Test
+ public void notActiveEvictionMAXStrategyTest() {
+ runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.MAX, maxResult);
+ }
+
+ @Test
+ public void notActiveEvictionMINStrategyTest() {
+ runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.MIN, minResult);
+ }
+
+ @Test
+ public void notActiveEvictionSUMStrategyTest() {
+ runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.SUM, sumResult);
+ }
+
+ @Test
+ public void notActiveEvictionPRIORITYStrategyTest() {
+ runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.PRIORITY, priorityResult);
+ }
+
+ /*
+ * Test cases for active policies
+ */
+
+ @Test
+ public void activeEvictionMAXStrategyTest() {
+ runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.MAX, maxResult);
+ }
+
+ @Test
+ public void activeEvictionMINStrategyTest() {
+ runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.MIN, minResult);
+ }
+
+ @Test
+ public void activeEvictionSUMStrategyTest() {
+ runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.SUM, sumResult);
+ }
+
+ @Test
+ public void activeEvictionPRIORITYStrategyTest() {
+ runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.PRIORITY, priorityResult);
+ }
+
+ /**
+ * Helper method: It runs the test with the given input using the not active
+ * policies and applies the strategy defined in the parameter.
+ *
+ * @param strategy
+ * the eviction strategy to be used
+ * @param expectedResult
+ * the result we expect
+ */
+ private void runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy strategy,
+ Integer[] expectedResult) {
+ @SuppressWarnings("unchecked")
+ MultiEvictionPolicy<Integer> multiEviction = new MultiEvictionPolicy<Integer>(strategy,
+ evictionPolicy1.clone(), evictionPolicy2.clone(), evictionPolicy3.clone());
+
+ List<Integer> result = new LinkedList<Integer>();
+
+ int buffersize = 0;
+ for (Integer tuple : tuples) {
+ // The buffer size should not matter, but we keep it for the case of
+ // later policy changes.
+ // The trigger does not matter. Always set it to false.
+ int eviction = multiEviction.notifyEviction(tuple, false, buffersize);
+ buffersize -= eviction;
+ result.add(eviction);
+
+ if (buffersize < 0) {
+ buffersize = 0;
+ }
+
+ buffersize++;
+ }
+
+ arrayEqualityCheck(expectedResult, result.toArray());
+ }
+
+ /**
+ * Helper method: It runs the test with the given input using the active
+ * policies and applies the strategy defined in the parameter.
+ *
+ * @param strategy
+ * the eviction strategy to be used
+ * @param expectedResult
+ * the result we expect
+ */
+ private void runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy strategy,
+ Integer[] expectedResult) {
+ @SuppressWarnings("unchecked")
+ MultiEvictionPolicy<Integer> multiEviction = new MultiEvictionPolicy<Integer>(strategy,
+ activeEvictionPolicy1.clone(), activeEvictionPolicy2.clone(),
+ activeEvictionPolicy3.clone());
+
+ List<Integer> result = new LinkedList<Integer>();
+
+ int buffersize = 0;
+ for (Integer tuple : tuples) {
+ // The buffer size should not matter, but we keep it for the case of
+ // later policy changes.
+ // The trigger does not matter. Always set it to false.
+ int eviction = multiEviction.notifyEvictionWithFakeElement(tuple, buffersize);
+ buffersize -= eviction;
+ result.add(eviction);
+
+ if (buffersize < 0) {
+ buffersize = 0;
+ }
+
+ buffersize++;
+ }
+
+ arrayEqualityCheck(expectedResult, result.toArray());
+ }
+
+ private void arrayEqualityCheck(Object[] array1, Object[] array2) {
+ assertEquals(
+ "The result arrays must have the same length. (Expected: " + Arrays.asList(array1)
+ + "; Actual: " + Arrays.asList(array2) + ")", array1.length, array2.length);
+ for (int i = 0; i < array1.length; i++) {
+ assertEquals("Unequal fields at position " + i + "(Expected: " + Arrays.asList(array1)
+ + "; Actual: " + Arrays.asList(array2) + ")", array1[i], array2[i]);
+ }
+ }
+
+}