You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/09/30 06:36:51 UTC
[GitHub] [flink-ml] gaoyunhaii opened a new pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
gaoyunhaii opened a new pull request #14:
URL: https://github.com/apache/flink-ml/pull/14
This PR adds the per-round operator wrappers. It would re-create new operator for different rounds, and isolate the states used by the operators.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] gaoyunhaii commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739765047
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
##########
@@ -0,0 +1,482 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.AbstractWrapperOperator;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.proxy.state.ProxyStateSnapshotContext;
+import org.apache.flink.iteration.proxy.state.ProxyStreamOperatorStateContext;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactoryUtil;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.util.LatencyStats;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+/** The base class for all the per-round wrapper operators. */
+public abstract class AbstractPerRoundWrapperOperator<T, S extends StreamOperator<T>>
+ extends AbstractWrapperOperator<T>
+ implements StreamOperatorStateHandler.CheckpointedStreamOperator {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractPerRoundWrapperOperator.class);
+
+ /** The wrapped operators for each round. */
+ private final Map<Integer, S> wrappedOperators;
+
+ protected final LatencyStats latencyStats;
+
+ private transient StreamOperatorStateContext streamOperatorStateContext;
+
+ private transient StreamOperatorStateHandler stateHandler;
+
+ private transient InternalTimeServiceManager<?> timeServiceManager;
+
+ private transient KeySelector<?, ?> stateKeySelector1;
+
+ private transient KeySelector<?, ?> stateKeySelector2;
+
+ public AbstractPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<T>> parameters,
+ StreamOperatorFactory<T> operatorFactory) {
+ super(parameters, operatorFactory);
+
+ this.wrappedOperators = new HashMap<>();
+ this.latencyStats = initializeLatencyStats();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ protected S getWrappedOperator(int round) {
+ S wrappedOperator = wrappedOperators.get(round);
+ if (wrappedOperator != null) {
+ return wrappedOperator;
+ }
+
+ // We needs to clone the operator factory to also support SimpleOperatorFactory.
+ try {
+ StreamOperatorFactory<T> clonedOperatorFactory =
+ InstantiationUtil.clone(operatorFactory);
+ wrappedOperator =
+ (S)
+ StreamOperatorFactoryUtil.<T, S>createOperator(
+ clonedOperatorFactory,
+ (StreamTask) parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ proxyOutput,
+ parameters.getOperatorEventDispatcher())
+ .f0;
+ initializeStreamOperator(wrappedOperator, round);
+ wrappedOperators.put(round, wrappedOperator);
+ return wrappedOperator;
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+
+ return wrappedOperator;
+ }
+
+ @Override
+ public void onEpochWatermarkIncrement(int epochWatermark) throws IOException {
+ try {
+ // Deserts all the operators with round < epoch watermark. Notes that
+ // the onEpochWatermarkIncrement must be from 0 and increment by 1 each time.
+ if (wrappedOperators.containsKey(epochWatermark)) {
+ closeStreamOperator(wrappedOperators.get(epochWatermark), epochWatermark);
+ wrappedOperators.remove(epochWatermark);
+ }
+
+ super.onEpochWatermarkIncrement(epochWatermark);
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+ }
+
+ protected void processForEachWrappedOperator(
+ BiConsumerWithException<Integer, S, Exception> consumer) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ consumer.accept(entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Override
+ public void open() throws Exception {}
+
+ @Override
+ public void initializeState(StreamTaskStateInitializer streamTaskStateManager)
+ throws Exception {
+ final TypeSerializer<?> keySerializer =
+ streamConfig.getStateKeySerializer(containingTask.getUserCodeClassLoader());
+
+ streamOperatorStateContext =
+ streamTaskStateManager.streamOperatorStateContext(
+ getOperatorID(),
+ getClass().getSimpleName(),
+ parameters.getProcessingTimeService(),
+ this,
+ keySerializer,
+ containingTask.getCancelables(),
+ metrics,
+ streamConfig.getManagedMemoryFractionOperatorUseCaseOfSlot(
+ ManagedMemoryUseCase.STATE_BACKEND,
+ containingTask
+ .getEnvironment()
+ .getTaskManagerInfo()
+ .getConfiguration(),
+ containingTask.getUserCodeClassLoader()),
+ isUsingCustomRawKeyedState());
+
+ stateHandler =
+ new StreamOperatorStateHandler(
+ streamOperatorStateContext,
+ containingTask.getExecutionConfig(),
+ containingTask.getCancelables());
+ stateHandler.initializeOperatorState(this);
+ this.timeServiceManager = streamOperatorStateContext.internalTimerServiceManager();
+
+ stateKeySelector1 =
+ streamConfig.getStatePartitioner(0, containingTask.getUserCodeClassLoader());
+ stateKeySelector2 =
+ streamConfig.getStatePartitioner(1, containingTask.getUserCodeClassLoader());
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ // Do thing for now since we do not have states.
+ }
+
+ @Internal
+ protected boolean isUsingCustomRawKeyedState() {
+ return false;
+ }
+
+ @Override
+ public void finish() throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ closeStreamOperator(entry.getValue(), entry.getKey());
+ }
+ wrappedOperators.clear();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (stateHandler != null) {
+ stateHandler.dispose();
+ }
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().prepareSnapshotPreBarrier(checkpointId);
+ }
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory factory)
+ throws Exception {
+ return stateHandler.snapshotState(
+ this,
+ Optional.ofNullable(timeServiceManager),
+ streamConfig.getOperatorName(),
+ checkpointId,
+ timestamp,
+ checkpointOptions,
+ factory,
+ isUsingCustomRawKeyedState());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ if (StreamOperatorStateHandler.CheckpointedStreamOperator.class.isAssignableFrom(
+ entry.getValue().getClass())) {
+ ((StreamOperatorStateHandler.CheckpointedStreamOperator) entry.getValue())
+ .snapshotState(new ProxyStateSnapshotContext(context));
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement1(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector1);
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement2(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector2);
+ }
+
+ private <T> void setKeyContextElement(StreamRecord<T> record, KeySelector<T, ?> selector)
+ throws Exception {
+ if (selector != null) {
+ Object key = selector.getKey(record.getValue());
+ setCurrentKey(key);
+ }
+ }
+
+ @Override
+ public OperatorMetricGroup getMetricGroup() {
+ return metrics;
+ }
+
+ @Override
+ public OperatorID getOperatorID() {
+ return streamConfig.getOperatorID();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long l) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().notifyCheckpointComplete(l);
+ }
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().notifyCheckpointAborted(checkpointId);
+ }
+ }
+
+ @Override
+ public void setCurrentKey(Object key) {
+ stateHandler.setCurrentKey(key);
+ }
+
+ @Override
+ public Object getCurrentKey() {
+ if (stateHandler == null) {
+ return null;
+ }
+
+ return stateHandler.getKeyedStateStore().orElse(null);
+ }
+
+ protected void reportOrForwardLatencyMarker(LatencyMarker marker) {
+ // all operators are tracking latencies
+ this.latencyStats.reportLatency(marker);
+
+ // everything except sinks forwards latency markers
+ this.output.emitLatencyMarker(marker);
+ }
+
+ private LatencyStats initializeLatencyStats() {
+ try {
+ Configuration taskManagerConfig =
+ containingTask.getEnvironment().getTaskManagerInfo().getConfiguration();
+ int historySize = taskManagerConfig.getInteger(MetricOptions.LATENCY_HISTORY_SIZE);
+ if (historySize <= 0) {
+ LOG.warn(
+ "{} has been set to a value equal or below 0: {}. Using default.",
+ MetricOptions.LATENCY_HISTORY_SIZE,
+ historySize);
+ historySize = MetricOptions.LATENCY_HISTORY_SIZE.defaultValue();
+ }
+
+ final String configuredGranularity =
+ taskManagerConfig.getString(MetricOptions.LATENCY_SOURCE_GRANULARITY);
+ LatencyStats.Granularity granularity;
+ try {
+ granularity =
+ LatencyStats.Granularity.valueOf(
+ configuredGranularity.toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException iae) {
+ granularity = LatencyStats.Granularity.OPERATOR;
+ LOG.warn(
+ "Configured value {} option for {} is invalid. Defaulting to {}.",
+ configuredGranularity,
+ MetricOptions.LATENCY_SOURCE_GRANULARITY.key(),
+ granularity);
+ }
+ MetricGroup jobMetricGroup = this.metrics.getJobMetricGroup();
+ return new LatencyStats(
+ jobMetricGroup.addGroup("latency"),
+ historySize,
+ containingTask.getIndexInSubtaskGroup(),
+ getOperatorID(),
+ granularity);
+ } catch (Exception e) {
+ LOG.warn("An error occurred while instantiating latency metrics.", e);
+ return new LatencyStats(
+ UnregisteredMetricGroups.createUnregisteredTaskManagerJobMetricGroup()
+ .addGroup("latency"),
+ 1,
+ 0,
+ new OperatorID(),
+ LatencyStats.Granularity.SINGLE);
+ }
+ }
+
+ private void initializeStreamOperator(S operator, int round) throws Exception {
+ operator.initializeState(
+ (operatorID,
+ operatorClassName,
+ processingTimeService,
+ keyContext,
+ keySerializer,
+ streamTaskCloseableRegistry,
+ metricGroup,
+ managedMemoryFraction,
+ isUsingCustomRawKeyedState) ->
+ new ProxyStreamOperatorStateContext(
+ streamOperatorStateContext, getRoundStatePrefix(round)));
+ operator.open();
+ }
+
+ private void closeStreamOperator(S operator, int round) throws Exception {
+ OperatorUtils.processOperatorOrUdfIfSatisfy(
+ operator, BoundedOneInput.class, BoundedOneInput::endInput);
+ operator.finish();
+ operator.close();
+
+ // Cleanup the states used by this operator.
+ cleanupOperatorStates(round);
+
+ if (stateHandler.getKeyedStateBackend() != null) {
+ cleanupKeyedStates(round);
+ }
+ }
+
+ private void cleanupOperatorStates(int round) {
+ String roundPrefix = getRoundStatePrefix(round);
+ OperatorStateBackend operatorStateBackend = stateHandler.getOperatorStateBackend();
+
+ if (operatorStateBackend instanceof DefaultOperatorStateBackend) {
+ for (String fieldNames :
+ new String[] {
+ "registeredOperatorStates",
+ "registeredBroadcastStates",
+ "accessedStatesByName",
+ "accessedBroadcastStatesByName"
+ }) {
+ Map<String, ?> field =
+ ReflectionUtils.getFieldValue(
+ operatorStateBackend,
+ DefaultOperatorStateBackend.class,
+ fieldNames);
+ field.entrySet().removeIf(entry -> entry.getKey().startsWith(roundPrefix));
+ }
+ } else {
+ LOG.warn("Unable to cleanup the operator state {}", operatorStateBackend);
+ }
+ }
+
+ private void cleanupKeyedStates(int round) {
+ String roundPrefix = getRoundStatePrefix(round);
Review comment:
Currently there should be no direct method to remove a state from the statebackend. I think for statebackends with no special treatment, we would first not support the cleanup during the iteration. I'll open a jira to continue tracking this issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r725957637
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/proxy/state/ProxyOperatorStateBackend.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ml.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.BroadcastState;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.SnapshotResult;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.RunnableFuture;
+
+/** Proxy {@link OperatorStateBackend} for the wrapped Operator. */
+public class ProxyOperatorStateBackend implements OperatorStateBackend {
+
+ private final OperatorStateBackend wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyOperatorStateBackend(OperatorStateBackend wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public <K, V> BroadcastState<K, V> getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)
+ throws Exception {
+ MapStateDescriptor<K, V> newDescriptor =
+ new MapStateDescriptor<>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getKeySerializer(),
+ stateDescriptor.getValueSerializer());
+ return wrappedBackend.getBroadcastState(newDescriptor);
+ }
+
+ @Override
+ public <S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ ListStateDescriptor<S> newDescriptor =
+ new ListStateDescriptor<>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getElementSerializer());
+ return wrappedBackend.getListState(newDescriptor);
+ }
+
+ @Override
+ public <S> ListState<S> getUnionListState(ListStateDescriptor<S> stateDescriptor)
+ throws Exception {
+ ListStateDescriptor<S> newDescriptor =
+ new ListStateDescriptor<S>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getElementSerializer());
+ return wrappedBackend.getUnionListState(stateDescriptor);
Review comment:
guess it should be `return wrappedBackend.getUnionListState(newDescriptor);`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] gaoyunhaii commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739765316
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/proxy/state/ProxyKeyedStateBackend.java
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateFunction;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.PriorityComparable;
+import org.apache.flink.runtime.state.SavepointResources;
+import org.apache.flink.runtime.state.SnapshotResult;
+import org.apache.flink.runtime.state.StateSnapshotTransformer;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.RunnableFuture;
+import java.util.stream.Stream;
+
+/** Proxy {@link KeyedStateBackend} for the wrapped operators. */
+public class ProxyKeyedStateBackend<K> implements CheckpointableKeyedStateBackend<K> {
+
+ private final CheckpointableKeyedStateBackend<K> wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyKeyedStateBackend(
+ CheckpointableKeyedStateBackend<K> wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public void setCurrentKey(K newKey) {
+ wrappedBackend.setCurrentKey(newKey);
+ }
+
+ @Override
+ public K getCurrentKey() {
+ return wrappedBackend.getCurrentKey();
+ }
+
+ @Override
+ public TypeSerializer<K> getKeySerializer() {
+ return wrappedBackend.getKeySerializer();
+ }
+
+ @Override
+ public <N, S extends State, T> void applyToAllKeys(
+ N namespace,
+ TypeSerializer<N> namespaceSerializer,
+ StateDescriptor<S, T> stateDescriptor,
+ KeyedStateFunction<K, S> function)
+ throws Exception {
+ StateDescriptor<S, T> newDescriptor = createNewDescriptor(stateDescriptor);
+ wrappedBackend.applyToAllKeys(namespace, namespaceSerializer, newDescriptor, function);
+ }
+
+ @Override
+ public <N> Stream<K> getKeys(String state, N namespace) {
Review comment:
Currently we could view the `namespace` as a sub-concept for each state, like for the WindowOperator, it represents the window id.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739746923
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/proxy/state/ProxyOperatorStateBackend.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.BroadcastState;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.SnapshotResult;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.RunnableFuture;
+
+/** Proxy {@link OperatorStateBackend} for the wrapped Operator. */
+public class ProxyOperatorStateBackend implements OperatorStateBackend {
+
+ private final OperatorStateBackend wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyOperatorStateBackend(OperatorStateBackend wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public <K, V> BroadcastState<K, V> getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)
+ throws Exception {
+ MapStateDescriptor<K, V> newDescriptor =
+ new MapStateDescriptor<>(
+ stateNamePrefix + stateDescriptor.getName(),
Review comment:
I think it might be more trackable that make a `stateNamePrefix + stateDescriptor.getName()` common method. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747431
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/proxy/state/ProxyKeyedStateBackend.java
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateFunction;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.PriorityComparable;
+import org.apache.flink.runtime.state.SavepointResources;
+import org.apache.flink.runtime.state.SnapshotResult;
+import org.apache.flink.runtime.state.StateSnapshotTransformer;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.RunnableFuture;
+import java.util.stream.Stream;
+
+/** Proxy {@link KeyedStateBackend} for the wrapped operators. */
+public class ProxyKeyedStateBackend<K> implements CheckpointableKeyedStateBackend<K> {
+
+ private final CheckpointableKeyedStateBackend<K> wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyKeyedStateBackend(
+ CheckpointableKeyedStateBackend<K> wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public void setCurrentKey(K newKey) {
+ wrappedBackend.setCurrentKey(newKey);
+ }
+
+ @Override
+ public K getCurrentKey() {
+ return wrappedBackend.getCurrentKey();
+ }
+
+ @Override
+ public TypeSerializer<K> getKeySerializer() {
+ return wrappedBackend.getKeySerializer();
+ }
+
+ @Override
+ public <N, S extends State, T> void applyToAllKeys(
+ N namespace,
Review comment:
Why does not the `namespace` have the prefix?
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/proxy/state/ProxyKeyedStateBackend.java
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateFunction;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.PriorityComparable;
+import org.apache.flink.runtime.state.SavepointResources;
+import org.apache.flink.runtime.state.SnapshotResult;
+import org.apache.flink.runtime.state.StateSnapshotTransformer;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.RunnableFuture;
+import java.util.stream.Stream;
+
+/** Proxy {@link KeyedStateBackend} for the wrapped operators. */
+public class ProxyKeyedStateBackend<K> implements CheckpointableKeyedStateBackend<K> {
+
+ private final CheckpointableKeyedStateBackend<K> wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyKeyedStateBackend(
+ CheckpointableKeyedStateBackend<K> wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public void setCurrentKey(K newKey) {
+ wrappedBackend.setCurrentKey(newKey);
+ }
+
+ @Override
+ public K getCurrentKey() {
+ return wrappedBackend.getCurrentKey();
+ }
+
+ @Override
+ public TypeSerializer<K> getKeySerializer() {
+ return wrappedBackend.getKeySerializer();
+ }
+
+ @Override
+ public <N, S extends State, T> void applyToAllKeys(
+ N namespace,
Review comment:
Why does not the `namespace` have the prefix?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-24652][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739805858
##########
File path: pom.xml
##########
@@ -18,726 +18,730 @@ specific language governing permissions and limitations
under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache</groupId>
- <artifactId>apache</artifactId>
- <version>23</version>
- </parent>
-
- <artifactId>flink-ml-parent</artifactId>
- <groupId>org.apache.flink</groupId>
- <name>Flink ML</name>
- <version>0.1-SNAPSHOT</version>
- <packaging>pom</packaging>
-
- <url>http://flink.apache.org</url>
- <inceptionYear>2019</inceptionYear>
-
- <licenses>
- <license>
- <name>The Apache Software License, Version 2.0</name>
- <url>https://www.apache.org/licenses/LICENSE-2.0.txt</url>
- <distribution>repo</distribution>
- </license>
- </licenses>
-
- <scm>
- <url>https://github.com/apache/flink-ml</url>
- <connection>git@github.com:apache/flink-ml.git</connection>
- <developerConnection>scm:git:https://gitbox.apache.org/repos/asf/flink-ml.git</developerConnection>
- </scm>
-
- <modules>
- <module>flink-ml-api</module>
- <module>flink-ml-lib</module>
- <module>flink-ml-uber</module>
- <module>flink-ml-iteration</module>
- <module>flink-ml-examples</module>
- </modules>
-
- <properties>
- <flink.shaded.version>12.0</flink.shaded.version>
- <scala.binary.version>2.11</scala.binary.version>
- <jackson.version>2.10.1</jackson.version>
- <target.java.version>1.8</target.java.version>
- <spotless.version>2.4.2</spotless.version>
- <slf4j.version>1.7.15</slf4j.version>
- <log4j.version>2.12.1</log4j.version>
- <junit.version>4.12</junit.version>
- <flink.forkCount>1C</flink.forkCount>
- <flink.reuseForks>true</flink.reuseForks>
- <flink.version>1.14.0</flink.version>
- <zookeeper.version>3.4.14</zookeeper.version>
-
- <!-- Can be set to any value to reproduce a specific build. -->
- <test.randomization.seed/>
- <test.unit.pattern>**/*Test.*</test.unit.pattern>
- </properties>
-
-
- <dependencies>
- <!-- Logging API -->
-
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </dependency>
-
- <!-- test dependencies -->
-
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <type>jar</type>
- <scope>test</scope>
- </dependency>
-
- <!-- tests will have log4j as the default logging framework available -->
-
- <dependency>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-slf4j-impl</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-api</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-core</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <!-- API bridge between log4j 1 and 2 -->
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-1.2-api</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
-
- <dependencyManagement>
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
Review comment:
Format？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747517
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/TwoInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,233 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+import org.apache.flink.streaming.runtime.tasks.TwoInputStreamTask;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputPerRoundWrapperOperator}. */
+public class TwoInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
Review comment:
Could be final.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747564
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/TwoInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,233 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+import org.apache.flink.streaming.runtime.tasks.TwoInputStreamTask;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputPerRoundWrapperOperator}. */
+public class TwoInputPerRoundWrapperOperatorTest {
Review comment:
Extend `TestLogger`
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/TwoInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,233 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+import org.apache.flink.streaming.runtime.tasks.TwoInputStreamTask;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputPerRoundWrapperOperator}. */
+public class TwoInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
+
+ @Test
+ public void testProcessElementsAndEpochWatermarks() throws Exception {
+ StreamOperatorFactory<IterationRecord<Integer>> wrapperFactory =
+ new WrapperOperatorFactory<>(
+ SimpleOperatorFactory.of(new LifeCycleTrackingTwoInputStreamOperator()),
+ new PerRoundOperatorWrapper<>());
+ OperatorID operatorId = new OperatorID();
+
+ try (StreamTaskMailboxTestHarness<IterationRecord<Integer>> harness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ TwoInputStreamTask::new,
+ new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .setupOutputForSingletonOperatorChain(wrapperFactory, operatorId)
+ .build()) {
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(5, 1), 2), 0);
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(6, 2), 3), 1);
+
+ // Check the output
Review comment:
Checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747090
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.operator.allround.OneInputAllRoundWrapperOperator;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractInput;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2;
+import org.apache.flink.streaming.api.operators.Input;
+import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.MultipleInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputAllRoundWrapperOperator}. */
+public class MultipleInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
Review comment:
Could be final
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r725964356
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/proxy/state/ProxyOperatorStateBackend.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ml.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.BroadcastState;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.SnapshotResult;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.RunnableFuture;
+
+/** Proxy {@link OperatorStateBackend} for the wrapped Operator. */
+public class ProxyOperatorStateBackend implements OperatorStateBackend {
+
+ private final OperatorStateBackend wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyOperatorStateBackend(OperatorStateBackend wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public <K, V> BroadcastState<K, V> getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)
+ throws Exception {
+ MapStateDescriptor<K, V> newDescriptor =
+ new MapStateDescriptor<>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getKeySerializer(),
+ stateDescriptor.getValueSerializer());
+ return wrappedBackend.getBroadcastState(newDescriptor);
+ }
+
+ @Override
+ public <S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ ListStateDescriptor<S> newDescriptor =
+ new ListStateDescriptor<>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getElementSerializer());
+ return wrappedBackend.getListState(newDescriptor);
+ }
+
+ @Override
+ public <S> ListState<S> getUnionListState(ListStateDescriptor<S> stateDescriptor)
+ throws Exception {
+ ListStateDescriptor<S> newDescriptor =
+ new ListStateDescriptor<S>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getElementSerializer());
+ return wrappedBackend.getUnionListState(stateDescriptor);
+ }
+
+ @Override
+ public Set<String> getRegisteredStateNames() {
+ Set<String> filteredNames = new HashSet<>();
+ Set<String> names = wrappedBackend.getRegisteredStateNames();
+
+ for (String name : names) {
+ if (name.startsWith(stateNamePrefix)) {
Review comment:
All states added by this proxy state backend will have its name starting with `stateNamePrefix`, but the reverse might not be true.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747467
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/proxy/state/ProxyKeyedStateBackend.java
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateFunction;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.PriorityComparable;
+import org.apache.flink.runtime.state.SavepointResources;
+import org.apache.flink.runtime.state.SnapshotResult;
+import org.apache.flink.runtime.state.StateSnapshotTransformer;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.RunnableFuture;
+import java.util.stream.Stream;
+
+/** Proxy {@link KeyedStateBackend} for the wrapped operators. */
+public class ProxyKeyedStateBackend<K> implements CheckpointableKeyedStateBackend<K> {
+
+ private final CheckpointableKeyedStateBackend<K> wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyKeyedStateBackend(
+ CheckpointableKeyedStateBackend<K> wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public void setCurrentKey(K newKey) {
+ wrappedBackend.setCurrentKey(newKey);
+ }
+
+ @Override
+ public K getCurrentKey() {
+ return wrappedBackend.getCurrentKey();
+ }
+
+ @Override
+ public TypeSerializer<K> getKeySerializer() {
+ return wrappedBackend.getKeySerializer();
+ }
+
+ @Override
+ public <N, S extends State, T> void applyToAllKeys(
+ N namespace,
+ TypeSerializer<N> namespaceSerializer,
+ StateDescriptor<S, T> stateDescriptor,
+ KeyedStateFunction<K, S> function)
+ throws Exception {
+ StateDescriptor<S, T> newDescriptor = createNewDescriptor(stateDescriptor);
+ wrappedBackend.applyToAllKeys(namespace, namespaceSerializer, newDescriptor, function);
+ }
+
+ @Override
+ public <N> Stream<K> getKeys(String state, N namespace) {
+ return wrappedBackend.getKeys(stateNamePrefix + state, namespace);
+ }
+
+ @Override
+ public <N> Stream<Tuple2<K, N>> getKeysAndNamespaces(String state) {
+ return wrappedBackend.getKeysAndNamespaces(stateNamePrefix + state);
+ }
+
+ @Override
+ public <N, S extends State, T> S getOrCreateKeyedState(
+ TypeSerializer<N> namespaceSerializer, StateDescriptor<S, T> stateDescriptor)
+ throws Exception {
+ StateDescriptor<S, T> newDescriptor = createNewDescriptor(stateDescriptor);
+ return wrappedBackend.getOrCreateKeyedState(namespaceSerializer, newDescriptor);
+ }
+
+ @Override
+ public <N, S extends State> S getPartitionedState(
+ N namespace,
+ TypeSerializer<N> namespaceSerializer,
+ StateDescriptor<S, ?> stateDescriptor)
+ throws Exception {
+ StateDescriptor<S, ?> newDescriptor = createNewDescriptor(stateDescriptor);
+ return wrappedBackend.getPartitionedState(namespace, namespaceSerializer, newDescriptor);
+ }
+
+ @Override
+ public void registerKeySelectionListener(KeySelectionListener<K> listener) {
+ wrappedBackend.registerKeySelectionListener(listener);
+ }
+
+ @Override
+ public boolean deregisterKeySelectionListener(KeySelectionListener<K> listener) {
+ return wrappedBackend.deregisterKeySelectionListener(listener);
+ }
+
+ @Nonnull
+ @Override
+ public <N, SV, SEV, S extends State, IS extends S> IS createInternalState(
+ @Nonnull TypeSerializer<N> namespaceSerializer,
+ @Nonnull StateDescriptor<S, SV> stateDesc,
+ @Nonnull
+ StateSnapshotTransformer.StateSnapshotTransformFactory<SEV>
+ snapshotTransformFactory)
+ throws Exception {
+ StateDescriptor<S, ?> newDescriptor = createNewDescriptor(stateDesc);
+ return wrappedBackend.createInternalState(
+ namespaceSerializer, newDescriptor, snapshotTransformFactory);
+ }
+
+ @SuppressWarnings("unchecked")
+ protected <S extends State, T> StateDescriptor<S, T> createNewDescriptor(
+ StateDescriptor<S, T> descriptor) {
+ switch (descriptor.getType()) {
+ case VALUE:
+ {
+ return (StateDescriptor<S, T>)
+ new ValueStateDescriptor<>(
+ stateNamePrefix + descriptor.getName(),
+ descriptor.getSerializer());
+ }
+ case LIST:
+ {
+ ListStateDescriptor<T> listStateDescriptor =
+ (ListStateDescriptor<T>) descriptor;
+ return (StateDescriptor<S, T>)
+ new ListStateDescriptor<>(
+ stateNamePrefix + listStateDescriptor.getName(),
+ listStateDescriptor.getElementSerializer());
+ }
+ case REDUCING:
+ {
+ ReducingStateDescriptor<T> reducingStateDescriptor =
+ (ReducingStateDescriptor<T>) descriptor;
+ return (StateDescriptor<S, T>)
+ new ReducingStateDescriptor<>(
+ stateNamePrefix + reducingStateDescriptor.getName(),
+ reducingStateDescriptor.getReduceFunction(),
+ reducingStateDescriptor.getSerializer());
+ }
+ case AGGREGATING:
+ {
+ AggregatingStateDescriptor<?, ?, T> aggregatingStateDescriptor =
+ (AggregatingStateDescriptor<?, ?, T>) descriptor;
+ return new AggregatingStateDescriptor(
+ stateNamePrefix + aggregatingStateDescriptor.getName(),
Review comment:
Same as above. Maybe we could extract a common method for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739746885
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/proxy/state/ProxyKeyedStateBackend.java
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateFunction;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.PriorityComparable;
+import org.apache.flink.runtime.state.SavepointResources;
+import org.apache.flink.runtime.state.SnapshotResult;
+import org.apache.flink.runtime.state.StateSnapshotTransformer;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.RunnableFuture;
+import java.util.stream.Stream;
+
+/** Proxy {@link KeyedStateBackend} for the wrapped operators. */
+public class ProxyKeyedStateBackend<K> implements CheckpointableKeyedStateBackend<K> {
+
+ private final CheckpointableKeyedStateBackend<K> wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyKeyedStateBackend(
+ CheckpointableKeyedStateBackend<K> wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public void setCurrentKey(K newKey) {
+ wrappedBackend.setCurrentKey(newKey);
+ }
+
+ @Override
+ public K getCurrentKey() {
+ return wrappedBackend.getCurrentKey();
+ }
+
+ @Override
+ public TypeSerializer<K> getKeySerializer() {
+ return wrappedBackend.getKeySerializer();
+ }
+
+ @Override
+ public <N, S extends State, T> void applyToAllKeys(
+ N namespace,
+ TypeSerializer<N> namespaceSerializer,
+ StateDescriptor<S, T> stateDescriptor,
+ KeyedStateFunction<K, S> function)
+ throws Exception {
+ StateDescriptor<S, T> newDescriptor = createNewDescriptor(stateDescriptor);
+ wrappedBackend.applyToAllKeys(namespace, namespaceSerializer, newDescriptor, function);
+ }
+
+ @Override
+ public <N> Stream<K> getKeys(String state, N namespace) {
Review comment:
Why the namespace don't have the prefix?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r735447020
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/PerRoundOperatorWrapper.java
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorWrapper;
+import org.apache.flink.iteration.proxy.ProxyKeySelector;
+import org.apache.flink.iteration.proxy.ProxyStreamPartitioner;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
+import org.apache.flink.util.OutputTag;
+
+/** The operator wrapper implementation for per-round wrappers. */
+public class PerRoundOperatorWrapper<T> implements OperatorWrapper<T, IterationRecord<T>> {
+
+ @Override
+ public StreamOperator<IterationRecord<T>> wrap(
Review comment:
I think we should give more detailed exception:
Tell the user that the iteration body could only build from the iteration variable and iteration constant.
BTW, maybe we should also update the java doc.
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
##########
@@ -0,0 +1,482 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.AbstractWrapperOperator;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.proxy.state.ProxyStateSnapshotContext;
+import org.apache.flink.iteration.proxy.state.ProxyStreamOperatorStateContext;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactoryUtil;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.util.LatencyStats;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+/** The base class for all the per-round wrapper operators. */
+public abstract class AbstractPerRoundWrapperOperator<T, S extends StreamOperator<T>>
+ extends AbstractWrapperOperator<T>
+ implements StreamOperatorStateHandler.CheckpointedStreamOperator {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractPerRoundWrapperOperator.class);
+
+ /** The wrapped operators for each round. */
+ private final Map<Integer, S> wrappedOperators;
+
+ protected final LatencyStats latencyStats;
+
+ private transient StreamOperatorStateContext streamOperatorStateContext;
+
+ private transient StreamOperatorStateHandler stateHandler;
+
+ private transient InternalTimeServiceManager<?> timeServiceManager;
+
+ private transient KeySelector<?, ?> stateKeySelector1;
+
+ private transient KeySelector<?, ?> stateKeySelector2;
+
+ public AbstractPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<T>> parameters,
+ StreamOperatorFactory<T> operatorFactory) {
+ super(parameters, operatorFactory);
+
+ this.wrappedOperators = new HashMap<>();
+ this.latencyStats = initializeLatencyStats();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ protected S getWrappedOperator(int round) {
+ S wrappedOperator = wrappedOperators.get(round);
+ if (wrappedOperator != null) {
+ return wrappedOperator;
+ }
+
+ // We needs to clone the operator factory to also support SimpleOperatorFactory.
+ try {
+ StreamOperatorFactory<T> clonedOperatorFactory =
+ InstantiationUtil.clone(operatorFactory);
+ wrappedOperator =
+ (S)
+ StreamOperatorFactoryUtil.<T, S>createOperator(
+ clonedOperatorFactory,
+ (StreamTask) parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ proxyOutput,
+ parameters.getOperatorEventDispatcher())
+ .f0;
+ initializeStreamOperator(wrappedOperator, round);
+ wrappedOperators.put(round, wrappedOperator);
+ return wrappedOperator;
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+
+ return wrappedOperator;
+ }
+
+ @Override
+ public void onEpochWatermarkIncrement(int epochWatermark) throws IOException {
+ try {
+ // Deserts all the operators with round < epoch watermark. Notes that
+ // the onEpochWatermarkIncrement must be from 0 and increment by 1 each time.
+ if (wrappedOperators.containsKey(epochWatermark)) {
+ closeStreamOperator(wrappedOperators.get(epochWatermark), epochWatermark);
+ wrappedOperators.remove(epochWatermark);
+ }
+
+ super.onEpochWatermarkIncrement(epochWatermark);
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+ }
+
+ protected void processForEachWrappedOperator(
+ BiConsumerWithException<Integer, S, Exception> consumer) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ consumer.accept(entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Override
+ public void open() throws Exception {}
+
+ @Override
+ public void initializeState(StreamTaskStateInitializer streamTaskStateManager)
+ throws Exception {
+ final TypeSerializer<?> keySerializer =
+ streamConfig.getStateKeySerializer(containingTask.getUserCodeClassLoader());
+
+ streamOperatorStateContext =
+ streamTaskStateManager.streamOperatorStateContext(
+ getOperatorID(),
+ getClass().getSimpleName(),
+ parameters.getProcessingTimeService(),
+ this,
+ keySerializer,
+ containingTask.getCancelables(),
+ metrics,
+ streamConfig.getManagedMemoryFractionOperatorUseCaseOfSlot(
+ ManagedMemoryUseCase.STATE_BACKEND,
+ containingTask
+ .getEnvironment()
+ .getTaskManagerInfo()
+ .getConfiguration(),
+ containingTask.getUserCodeClassLoader()),
+ isUsingCustomRawKeyedState());
+
+ stateHandler =
+ new StreamOperatorStateHandler(
+ streamOperatorStateContext,
+ containingTask.getExecutionConfig(),
+ containingTask.getCancelables());
+ stateHandler.initializeOperatorState(this);
+ this.timeServiceManager = streamOperatorStateContext.internalTimerServiceManager();
+
+ stateKeySelector1 =
+ streamConfig.getStatePartitioner(0, containingTask.getUserCodeClassLoader());
+ stateKeySelector2 =
+ streamConfig.getStatePartitioner(1, containingTask.getUserCodeClassLoader());
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ // Do thing for now since we do not have states.
+ }
+
+ @Internal
+ protected boolean isUsingCustomRawKeyedState() {
+ return false;
+ }
+
+ @Override
+ public void finish() throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ closeStreamOperator(entry.getValue(), entry.getKey());
+ }
+ wrappedOperators.clear();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (stateHandler != null) {
+ stateHandler.dispose();
+ }
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().prepareSnapshotPreBarrier(checkpointId);
+ }
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory factory)
+ throws Exception {
+ return stateHandler.snapshotState(
+ this,
+ Optional.ofNullable(timeServiceManager),
+ streamConfig.getOperatorName(),
+ checkpointId,
+ timestamp,
+ checkpointOptions,
+ factory,
+ isUsingCustomRawKeyedState());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ if (StreamOperatorStateHandler.CheckpointedStreamOperator.class.isAssignableFrom(
+ entry.getValue().getClass())) {
+ ((StreamOperatorStateHandler.CheckpointedStreamOperator) entry.getValue())
+ .snapshotState(new ProxyStateSnapshotContext(context));
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement1(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector1);
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement2(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector2);
+ }
+
+ private <T> void setKeyContextElement(StreamRecord<T> record, KeySelector<T, ?> selector)
+ throws Exception {
+ if (selector != null) {
+ Object key = selector.getKey(record.getValue());
+ setCurrentKey(key);
+ }
+ }
+
+ @Override
+ public OperatorMetricGroup getMetricGroup() {
+ return metrics;
+ }
+
+ @Override
+ public OperatorID getOperatorID() {
+ return streamConfig.getOperatorID();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long l) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().notifyCheckpointComplete(l);
+ }
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().notifyCheckpointAborted(checkpointId);
+ }
+ }
+
+ @Override
+ public void setCurrentKey(Object key) {
+ stateHandler.setCurrentKey(key);
+ }
+
+ @Override
+ public Object getCurrentKey() {
+ if (stateHandler == null) {
+ return null;
+ }
+
+ return stateHandler.getKeyedStateStore().orElse(null);
+ }
+
+ protected void reportOrForwardLatencyMarker(LatencyMarker marker) {
+ // all operators are tracking latencies
+ this.latencyStats.reportLatency(marker);
+
+ // everything except sinks forwards latency markers
+ this.output.emitLatencyMarker(marker);
+ }
+
+ private LatencyStats initializeLatencyStats() {
Review comment:
I am not sure that the exact meaning of latency marker in the iteration. But it seems that both the wrapper and "internal" operators(per-round) have this metrics. So maybe only one of them have the metrics and report it is enough. And the code that initializing the `latencymarker` is the same as the internal implementation, which might introduce some burden in the future.
IMHO an alternative way might be just report the limitation in the log. And after we could make it clear we add this back.
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
##########
@@ -0,0 +1,482 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.AbstractWrapperOperator;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.proxy.state.ProxyStateSnapshotContext;
+import org.apache.flink.iteration.proxy.state.ProxyStreamOperatorStateContext;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactoryUtil;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.util.LatencyStats;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+/** The base class for all the per-round wrapper operators. */
+public abstract class AbstractPerRoundWrapperOperator<T, S extends StreamOperator<T>>
+ extends AbstractWrapperOperator<T>
+ implements StreamOperatorStateHandler.CheckpointedStreamOperator {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractPerRoundWrapperOperator.class);
+
+ /** The wrapped operators for each round. */
+ private final Map<Integer, S> wrappedOperators;
+
+ protected final LatencyStats latencyStats;
+
+ private transient StreamOperatorStateContext streamOperatorStateContext;
+
+ private transient StreamOperatorStateHandler stateHandler;
+
+ private transient InternalTimeServiceManager<?> timeServiceManager;
+
+ private transient KeySelector<?, ?> stateKeySelector1;
+
+ private transient KeySelector<?, ?> stateKeySelector2;
+
+ public AbstractPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<T>> parameters,
+ StreamOperatorFactory<T> operatorFactory) {
+ super(parameters, operatorFactory);
+
+ this.wrappedOperators = new HashMap<>();
+ this.latencyStats = initializeLatencyStats();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ protected S getWrappedOperator(int round) {
+ S wrappedOperator = wrappedOperators.get(round);
+ if (wrappedOperator != null) {
+ return wrappedOperator;
+ }
+
+ // We needs to clone the operator factory to also support SimpleOperatorFactory.
+ try {
+ StreamOperatorFactory<T> clonedOperatorFactory =
+ InstantiationUtil.clone(operatorFactory);
+ wrappedOperator =
+ (S)
+ StreamOperatorFactoryUtil.<T, S>createOperator(
+ clonedOperatorFactory,
+ (StreamTask) parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ proxyOutput,
+ parameters.getOperatorEventDispatcher())
+ .f0;
+ initializeStreamOperator(wrappedOperator, round);
+ wrappedOperators.put(round, wrappedOperator);
+ return wrappedOperator;
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+
+ return wrappedOperator;
+ }
+
+ @Override
+ public void onEpochWatermarkIncrement(int epochWatermark) throws IOException {
+ try {
+ // Deserts all the operators with round < epoch watermark. Notes that
+ // the onEpochWatermarkIncrement must be from 0 and increment by 1 each time.
+ if (wrappedOperators.containsKey(epochWatermark)) {
+ closeStreamOperator(wrappedOperators.get(epochWatermark), epochWatermark);
+ wrappedOperators.remove(epochWatermark);
+ }
+
+ super.onEpochWatermarkIncrement(epochWatermark);
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+ }
+
+ protected void processForEachWrappedOperator(
+ BiConsumerWithException<Integer, S, Exception> consumer) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ consumer.accept(entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Override
+ public void open() throws Exception {}
+
+ @Override
+ public void initializeState(StreamTaskStateInitializer streamTaskStateManager)
+ throws Exception {
+ final TypeSerializer<?> keySerializer =
+ streamConfig.getStateKeySerializer(containingTask.getUserCodeClassLoader());
+
+ streamOperatorStateContext =
+ streamTaskStateManager.streamOperatorStateContext(
+ getOperatorID(),
+ getClass().getSimpleName(),
+ parameters.getProcessingTimeService(),
+ this,
+ keySerializer,
+ containingTask.getCancelables(),
+ metrics,
+ streamConfig.getManagedMemoryFractionOperatorUseCaseOfSlot(
+ ManagedMemoryUseCase.STATE_BACKEND,
+ containingTask
+ .getEnvironment()
+ .getTaskManagerInfo()
+ .getConfiguration(),
+ containingTask.getUserCodeClassLoader()),
+ isUsingCustomRawKeyedState());
+
+ stateHandler =
+ new StreamOperatorStateHandler(
+ streamOperatorStateContext,
+ containingTask.getExecutionConfig(),
+ containingTask.getCancelables());
+ stateHandler.initializeOperatorState(this);
+ this.timeServiceManager = streamOperatorStateContext.internalTimerServiceManager();
+
+ stateKeySelector1 =
+ streamConfig.getStatePartitioner(0, containingTask.getUserCodeClassLoader());
+ stateKeySelector2 =
+ streamConfig.getStatePartitioner(1, containingTask.getUserCodeClassLoader());
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ // Do thing for now since we do not have states.
+ }
+
+ @Internal
+ protected boolean isUsingCustomRawKeyedState() {
+ return false;
+ }
+
+ @Override
+ public void finish() throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ closeStreamOperator(entry.getValue(), entry.getKey());
+ }
+ wrappedOperators.clear();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (stateHandler != null) {
+ stateHandler.dispose();
+ }
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().prepareSnapshotPreBarrier(checkpointId);
+ }
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory factory)
+ throws Exception {
+ return stateHandler.snapshotState(
+ this,
+ Optional.ofNullable(timeServiceManager),
+ streamConfig.getOperatorName(),
+ checkpointId,
+ timestamp,
+ checkpointOptions,
+ factory,
+ isUsingCustomRawKeyedState());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ if (StreamOperatorStateHandler.CheckpointedStreamOperator.class.isAssignableFrom(
+ entry.getValue().getClass())) {
+ ((StreamOperatorStateHandler.CheckpointedStreamOperator) entry.getValue())
+ .snapshotState(new ProxyStateSnapshotContext(context));
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement1(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector1);
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement2(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector2);
+ }
+
+ private <T> void setKeyContextElement(StreamRecord<T> record, KeySelector<T, ?> selector)
+ throws Exception {
+ if (selector != null) {
+ Object key = selector.getKey(record.getValue());
+ setCurrentKey(key);
+ }
+ }
+
+ @Override
+ public OperatorMetricGroup getMetricGroup() {
+ return metrics;
+ }
+
+ @Override
+ public OperatorID getOperatorID() {
+ return streamConfig.getOperatorID();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long l) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().notifyCheckpointComplete(l);
+ }
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().notifyCheckpointAborted(checkpointId);
+ }
+ }
+
+ @Override
+ public void setCurrentKey(Object key) {
+ stateHandler.setCurrentKey(key);
+ }
+
+ @Override
+ public Object getCurrentKey() {
+ if (stateHandler == null) {
+ return null;
+ }
+
+ return stateHandler.getKeyedStateStore().orElse(null);
+ }
+
+ protected void reportOrForwardLatencyMarker(LatencyMarker marker) {
+ // all operators are tracking latencies
+ this.latencyStats.reportLatency(marker);
+
+ // everything except sinks forwards latency markers
+ this.output.emitLatencyMarker(marker);
+ }
+
+ private LatencyStats initializeLatencyStats() {
+ try {
+ Configuration taskManagerConfig =
+ containingTask.getEnvironment().getTaskManagerInfo().getConfiguration();
+ int historySize = taskManagerConfig.getInteger(MetricOptions.LATENCY_HISTORY_SIZE);
+ if (historySize <= 0) {
+ LOG.warn(
+ "{} has been set to a value equal or below 0: {}. Using default.",
+ MetricOptions.LATENCY_HISTORY_SIZE,
+ historySize);
+ historySize = MetricOptions.LATENCY_HISTORY_SIZE.defaultValue();
+ }
+
+ final String configuredGranularity =
+ taskManagerConfig.getString(MetricOptions.LATENCY_SOURCE_GRANULARITY);
+ LatencyStats.Granularity granularity;
+ try {
+ granularity =
+ LatencyStats.Granularity.valueOf(
+ configuredGranularity.toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException iae) {
+ granularity = LatencyStats.Granularity.OPERATOR;
+ LOG.warn(
+ "Configured value {} option for {} is invalid. Defaulting to {}.",
+ configuredGranularity,
+ MetricOptions.LATENCY_SOURCE_GRANULARITY.key(),
+ granularity);
+ }
+ MetricGroup jobMetricGroup = this.metrics.getJobMetricGroup();
+ return new LatencyStats(
+ jobMetricGroup.addGroup("latency"),
+ historySize,
+ containingTask.getIndexInSubtaskGroup(),
+ getOperatorID(),
+ granularity);
+ } catch (Exception e) {
+ LOG.warn("An error occurred while instantiating latency metrics.", e);
+ return new LatencyStats(
+ UnregisteredMetricGroups.createUnregisteredTaskManagerJobMetricGroup()
+ .addGroup("latency"),
+ 1,
+ 0,
+ new OperatorID(),
+ LatencyStats.Granularity.SINGLE);
+ }
+ }
+
+ private void initializeStreamOperator(S operator, int round) throws Exception {
+ operator.initializeState(
+ (operatorID,
+ operatorClassName,
+ processingTimeService,
+ keyContext,
+ keySerializer,
+ streamTaskCloseableRegistry,
+ metricGroup,
+ managedMemoryFraction,
+ isUsingCustomRawKeyedState) ->
+ new ProxyStreamOperatorStateContext(
+ streamOperatorStateContext, getRoundStatePrefix(round)));
+ operator.open();
+ }
+
+ private void closeStreamOperator(S operator, int round) throws Exception {
+ OperatorUtils.processOperatorOrUdfIfSatisfy(
+ operator, BoundedOneInput.class, BoundedOneInput::endInput);
+ operator.finish();
+ operator.close();
+
+ // Cleanup the states used by this operator.
+ cleanupOperatorStates(round);
+
+ if (stateHandler.getKeyedStateBackend() != null) {
+ cleanupKeyedStates(round);
+ }
+ }
+
+ private void cleanupOperatorStates(int round) {
+ String roundPrefix = getRoundStatePrefix(round);
+ OperatorStateBackend operatorStateBackend = stateHandler.getOperatorStateBackend();
+
+ if (operatorStateBackend instanceof DefaultOperatorStateBackend) {
+ for (String fieldNames :
+ new String[] {
+ "registeredOperatorStates",
+ "registeredBroadcastStates",
+ "accessedStatesByName",
+ "accessedBroadcastStatesByName"
+ }) {
+ Map<String, ?> field =
+ ReflectionUtils.getFieldValue(
+ operatorStateBackend,
+ DefaultOperatorStateBackend.class,
+ fieldNames);
+ field.entrySet().removeIf(entry -> entry.getKey().startsWith(roundPrefix));
+ }
+ } else {
+ LOG.warn("Unable to cleanup the operator state {}", operatorStateBackend);
+ }
+ }
+
+ private void cleanupKeyedStates(int round) {
+ String roundPrefix = getRoundStatePrefix(round);
Review comment:
I know that we should clean-up the per-round state. But it might be not so elegant that depending a specific implementation of the `statebackend`. Is there general way that could clean up the per-round's state.
Or maybe you could open a jira to tracking this improvements.
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/Iterations.java
##########
@@ -108,7 +108,7 @@ public static DataStreamList iterateUnboundedStreams(
public static DataStreamList iterateBoundedStreamsUntilTermination(
DataStreamList initVariableStreams, DataStreamList dataStreams, IterationBody body) {
return IterationFactory.createIteration(
- initVariableStreams, dataStreams, body, new AllRoundOperatorWrapper(), true);
+ initVariableStreams, dataStreams, body, new AllRoundOperatorWrapper(), true);
Review comment:
Maybe this is the extra spaces?
##########
File path: flink-ml-iteration/pom.xml
##########
@@ -61,6 +61,14 @@ under the License.
<scope>provided</scope>
</dependency>
+ <!-- We have special treatment with the rocksdb state. -->
Review comment:
Do you think that it is possible that we could only depend on the rocksdb in the testing phase?
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperator.java
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.streaming.api.graph.StreamEdge;
+import org.apache.flink.streaming.api.operators.Input;
+import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Per-round wrapper for the multiple-inputs operator. */
+public class MultipleInputPerRoundWrapperOperator<OUT>
+ extends AbstractPerRoundWrapperOperator<OUT, MultipleInputStreamOperator<OUT>>
+ implements MultipleInputStreamOperator<IterationRecord<OUT>> {
+
+ public MultipleInputPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<OUT>> parameters,
+ StreamOperatorFactory<OUT> operatorFactory) {
+ super(parameters, operatorFactory);
+ }
+
+ private <IN> void processElement(
+ int inputIndex,
+ Input<IN> input,
+ StreamRecord<IN> reusedInput,
+ StreamRecord<IterationRecord<IN>> element)
+ throws Exception {
+ switch (element.getValue().getType()) {
+ case RECORD:
+ reusedInput.replace(element.getValue().getValue(), element.getTimestamp());
+ setIterationContextRound(element.getValue().getEpoch());
+ input.processElement(reusedInput);
+ clearIterationContextRound();
+ break;
+ case EPOCH_WATERMARK:
+ onEpochWatermarkEvent(inputIndex, element.getValue());
+ break;
+ default:
+ throw new FlinkRuntimeException("Not supported iteration record type: " + element);
+ }
+ }
+
+ @Override
+ @SuppressWarnings({"rawtypes"})
+ public List<Input> getInputs() {
+ List<Input> proxyInputs = new ArrayList<>();
+
+ // Determine how much inputs we have
+ List<StreamEdge> inEdges =
+ streamConfig.getInPhysicalEdges(containingTask.getUserCodeClassLoader());
+ int numberOfInputs =
+ inEdges.stream().map(StreamEdge::getTypeNumber).collect(Collectors.toSet()).size();
+
+ for (int i = 0; i < numberOfInputs; ++i) {
+ // TODO: Note that here we relies on the assumption that the
Review comment:
Is there any way to assert this assumption?
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
##########
@@ -0,0 +1,482 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.AbstractWrapperOperator;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.proxy.state.ProxyStateSnapshotContext;
+import org.apache.flink.iteration.proxy.state.ProxyStreamOperatorStateContext;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactoryUtil;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.util.LatencyStats;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+/** The base class for all the per-round wrapper operators. */
+public abstract class AbstractPerRoundWrapperOperator<T, S extends StreamOperator<T>>
+ extends AbstractWrapperOperator<T>
+ implements StreamOperatorStateHandler.CheckpointedStreamOperator {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractPerRoundWrapperOperator.class);
+
+ /** The wrapped operators for each round. */
+ private final Map<Integer, S> wrappedOperators;
+
+ protected final LatencyStats latencyStats;
+
+ private transient StreamOperatorStateContext streamOperatorStateContext;
+
+ private transient StreamOperatorStateHandler stateHandler;
+
+ private transient InternalTimeServiceManager<?> timeServiceManager;
+
+ private transient KeySelector<?, ?> stateKeySelector1;
+
+ private transient KeySelector<?, ?> stateKeySelector2;
+
+ public AbstractPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<T>> parameters,
+ StreamOperatorFactory<T> operatorFactory) {
+ super(parameters, operatorFactory);
+
+ this.wrappedOperators = new HashMap<>();
+ this.latencyStats = initializeLatencyStats();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ protected S getWrappedOperator(int round) {
+ S wrappedOperator = wrappedOperators.get(round);
+ if (wrappedOperator != null) {
+ return wrappedOperator;
+ }
+
+ // We needs to clone the operator factory to also support SimpleOperatorFactory.
+ try {
+ StreamOperatorFactory<T> clonedOperatorFactory =
+ InstantiationUtil.clone(operatorFactory);
+ wrappedOperator =
+ (S)
+ StreamOperatorFactoryUtil.<T, S>createOperator(
+ clonedOperatorFactory,
+ (StreamTask) parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ proxyOutput,
+ parameters.getOperatorEventDispatcher())
+ .f0;
+ initializeStreamOperator(wrappedOperator, round);
+ wrappedOperators.put(round, wrappedOperator);
+ return wrappedOperator;
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+
+ return wrappedOperator;
+ }
+
+ @Override
+ public void onEpochWatermarkIncrement(int epochWatermark) throws IOException {
+ try {
+ // Deserts all the operators with round < epoch watermark. Notes that
Review comment:
Deserts? or Destroy.
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
##########
@@ -0,0 +1,482 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.AbstractWrapperOperator;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.proxy.state.ProxyStateSnapshotContext;
+import org.apache.flink.iteration.proxy.state.ProxyStreamOperatorStateContext;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactoryUtil;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.util.LatencyStats;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+/** The base class for all the per-round wrapper operators. */
+public abstract class AbstractPerRoundWrapperOperator<T, S extends StreamOperator<T>>
+ extends AbstractWrapperOperator<T>
+ implements StreamOperatorStateHandler.CheckpointedStreamOperator {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractPerRoundWrapperOperator.class);
+
+ /** The wrapped operators for each round. */
+ private final Map<Integer, S> wrappedOperators;
+
+ protected final LatencyStats latencyStats;
+
+ private transient StreamOperatorStateContext streamOperatorStateContext;
+
+ private transient StreamOperatorStateHandler stateHandler;
+
+ private transient InternalTimeServiceManager<?> timeServiceManager;
+
+ private transient KeySelector<?, ?> stateKeySelector1;
+
+ private transient KeySelector<?, ?> stateKeySelector2;
+
+ public AbstractPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<T>> parameters,
+ StreamOperatorFactory<T> operatorFactory) {
+ super(parameters, operatorFactory);
+
+ this.wrappedOperators = new HashMap<>();
+ this.latencyStats = initializeLatencyStats();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ protected S getWrappedOperator(int round) {
+ S wrappedOperator = wrappedOperators.get(round);
+ if (wrappedOperator != null) {
+ return wrappedOperator;
+ }
+
+ // We needs to clone the operator factory to also support SimpleOperatorFactory.
Review comment:
need?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] gaoyunhaii closed pull request #14: [FLINK-24652][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
gaoyunhaii closed pull request #14:
URL: https://github.com/apache/flink-ml/pull/14
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739746994
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.operator.allround.OneInputAllRoundWrapperOperator;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractInput;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2;
+import org.apache.flink.streaming.api.operators.Input;
+import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.MultipleInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputAllRoundWrapperOperator}. */
+public class MultipleInputPerRoundWrapperOperatorTest {
Review comment:
Extend `TestLogger`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] gaoyunhaii commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r731603315
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/proxy/state/ProxyOperatorStateBackend.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ml.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.BroadcastState;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.SnapshotResult;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.RunnableFuture;
+
+/** Proxy {@link OperatorStateBackend} for the wrapped Operator. */
+public class ProxyOperatorStateBackend implements OperatorStateBackend {
+
+ private final OperatorStateBackend wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyOperatorStateBackend(OperatorStateBackend wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public <K, V> BroadcastState<K, V> getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)
+ throws Exception {
+ MapStateDescriptor<K, V> newDescriptor =
+ new MapStateDescriptor<>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getKeySerializer(),
+ stateDescriptor.getValueSerializer());
+ return wrappedBackend.getBroadcastState(newDescriptor);
+ }
+
+ @Override
+ public <S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ ListStateDescriptor<S> newDescriptor =
+ new ListStateDescriptor<>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getElementSerializer());
+ return wrappedBackend.getListState(newDescriptor);
+ }
+
+ @Override
+ public <S> ListState<S> getUnionListState(ListStateDescriptor<S> stateDescriptor)
+ throws Exception {
+ ListStateDescriptor<S> newDescriptor =
+ new ListStateDescriptor<S>(
+ stateNamePrefix + stateDescriptor.getName(),
+ stateDescriptor.getElementSerializer());
+ return wrappedBackend.getUnionListState(stateDescriptor);
+ }
+
+ @Override
+ public Set<String> getRegisteredStateNames() {
+ Set<String> filteredNames = new HashSet<>();
+ Set<String> names = wrappedBackend.getRegisteredStateNames();
+
+ for (String name : names) {
+ if (name.startsWith(stateNamePrefix)) {
Review comment:
There should be no reverse cases here due to all the state operator would go across the proxy state backend, that means even if users have a state named `r5-xx` for round 6, it would be renamed to `r6-r5-xx`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747168
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.operator.allround.OneInputAllRoundWrapperOperator;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractInput;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2;
+import org.apache.flink.streaming.api.operators.Input;
+import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.MultipleInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputAllRoundWrapperOperator}. */
+public class MultipleInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
+
+ @Test
+ public void testProcessElementsAndEpochWatermarks() throws Exception {
+ StreamOperatorFactory<IterationRecord<Integer>> wrapperFactory =
+ new WrapperOperatorFactory<>(
+ new LifeCycleTrackingMultiInputStreamOperatorFactory(),
+ new PerRoundOperatorWrapper<>());
+ OperatorID operatorId = new OperatorID();
+
+ try (StreamTaskMailboxTestHarness<IterationRecord<Integer>> harness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ MultipleInputStreamTask::new,
+ new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .setupOutputForSingletonOperatorChain(wrapperFactory, operatorId)
+ .build()) {
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(5, 1), 2), 0);
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(6, 2), 3), 2);
+
+ // Check the output
+ assertEquals(
+ Arrays.asList(
+ new StreamRecord<>(IterationRecord.newRecord(5, 1), 2),
+ new StreamRecord<>(IterationRecord.newRecord(6, 2), 3)),
+ new ArrayList<>(harness.getOutput()));
+
+ // Check the other lifecycles.
+ harness.getStreamTask()
+ .triggerCheckpointOnBarrier(
+ new CheckpointMetaData(5, 2),
+ CheckpointOptions.alignedNoTimeout(
+ CheckpointType.CHECKPOINT,
+ CheckpointStorageLocationReference.getDefault()),
+ new CheckpointMetricsBuilder()
+ .setAlignmentDurationNanos(0)
+ .setBytesProcessedDuringAlignment(0));
+ harness.processAll();
+
+ harness.getStreamTask().notifyCheckpointCompleteAsync(5);
+ harness.processAll();
+
+ harness.getStreamTask().notifyCheckpointAbortAsync(6, 5);
+ harness.processAll();
+
+ harness.getOutput().clear();
+ harness.processElement(
+ new StreamRecord<>(IterationRecord.newEpochWatermark(1, "only-one-0")), 0);
+ harness.processElement(
+ new StreamRecord<>(IterationRecord.newEpochWatermark(1, "only-one-1")), 1);
+ harness.processElement(
+ new StreamRecord<>(IterationRecord.newEpochWatermark(1, "only-one-2")), 2);
+ harness.processElement(
+ new StreamRecord<>(IterationRecord.newEpochWatermark(2, "only-one-0")), 0);
+ harness.processElement(
+ new StreamRecord<>(IterationRecord.newEpochWatermark(2, "only-one-1")), 1);
+ harness.processElement(
+ new StreamRecord<>(IterationRecord.newEpochWatermark(2, "only-one-2")), 2);
+
+ // Check the output
+ assertEquals(
+ Arrays.asList(
+ new StreamRecord<>(
+ IterationRecord.newEpochWatermark(
+ 1, OperatorUtils.getUniqueSenderId(operatorId, 0))),
+ new StreamRecord<>(
+ IterationRecord.newEpochWatermark(
+ 2, OperatorUtils.getUniqueSenderId(operatorId, 0)))),
+ new ArrayList<>(harness.getOutput()));
+
+ harness.processEvent(EndOfData.INSTANCE, 0);
+ harness.processEvent(EndOfData.INSTANCE, 1);
+ harness.processEvent(EndOfData.INSTANCE, 2);
+ harness.endInput();
+ harness.finishProcessing();
+
+ assertEquals(
+ Arrays.asList(
+ /* First wrapped operator */
+ LifeCycle.INITIALIZE_STATE,
+ LifeCycle.OPEN,
+ LifeCycle.PROCESS_ELEMENT,
+ /* second wrapped operator */
+ LifeCycle.INITIALIZE_STATE,
+ LifeCycle.OPEN,
+ LifeCycle.PROCESS_ELEMENT,
+ /* states */
+ LifeCycle.PREPARE_SNAPSHOT_PRE_BARRIER,
+ LifeCycle.PREPARE_SNAPSHOT_PRE_BARRIER,
+ LifeCycle.SNAPSHOT_STATE,
+ LifeCycle.SNAPSHOT_STATE,
+ LifeCycle.NOTIFY_CHECKPOINT_COMPLETE,
+ LifeCycle.NOTIFY_CHECKPOINT_COMPLETE,
+ LifeCycle.NOTIFY_CHECKPOINT_ABORT,
+ LifeCycle.NOTIFY_CHECKPOINT_ABORT,
+ LifeCycle.FINISH,
+ LifeCycle.CLOSE,
+ LifeCycle.FINISH,
+ LifeCycle.CLOSE),
+ lifeCycles);
+ }
+ }
+
+ private static class LifeCycleTrackingMultiInputStreamOperator
+ extends AbstractStreamOperatorV2<Integer>
+ implements MultipleInputStreamOperator<Integer> {
+
+ private final int numberOfInputs;
+
+ public LifeCycleTrackingMultiInputStreamOperator(
+ StreamOperatorParameters<Integer> parameters, int numberOfInputs) {
+ super(parameters, numberOfInputs);
+ this.numberOfInputs = numberOfInputs;
+ }
+
+ @Override
+ public List<Input> getInputs() {
+ List<Input> inputs = new ArrayList<>();
+ for (int i = 0; i < numberOfInputs; ++i) {
+ inputs.add(
+ new AbstractInput(this, i + 1) {
+ @Override
+ public void processElement(StreamRecord element) throws Exception {
+ output.collect(element);
+ lifeCycles.add(LifeCycle.PROCESS_ELEMENT);
+ }
+ });
+ }
+
+ return inputs;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ lifeCycles.add(LifeCycle.OPEN);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ lifeCycles.add(LifeCycle.INITIALIZE_STATE);
+ }
+
+ @Override
+ public void finish() throws Exception {
+ super.finish();
+ lifeCycles.add(LifeCycle.FINISH);
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ lifeCycles.add(LifeCycle.CLOSE);
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ super.prepareSnapshotPreBarrier(checkpointId);
+ lifeCycles.add(LifeCycle.PREPARE_SNAPSHOT_PRE_BARRIER);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ lifeCycles.add(LifeCycle.SNAPSHOT_STATE);
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ super.notifyCheckpointComplete(checkpointId);
+ lifeCycles.add(LifeCycle.NOTIFY_CHECKPOINT_COMPLETE);
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ super.notifyCheckpointAborted(checkpointId);
+ lifeCycles.add(LifeCycle.NOTIFY_CHECKPOINT_ABORT);
+ }
+ }
+
+ /** Life-cycle tracking stream operator factory. */
+ public static class LifeCycleTrackingMultiInputStreamOperatorFactory
Review comment:
Does this need to be `public`? I do not see it used other than this class.
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputPerRoundWrapperOperator}. */
+public class OneInputPerRoundWrapperOperatorTest {
Review comment:
Extend `TestLogger`
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputPerRoundWrapperOperator}. */
+public class OneInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
Review comment:
Could be `final`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747189
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputPerRoundWrapperOperator}. */
+public class OneInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
+
+ @Test
+ public void testProcessElementsAndEpochWatermarks() throws Exception {
+ StreamOperatorFactory<IterationRecord<Integer>> wrapperFactory =
+ new WrapperOperatorFactory<>(
+ SimpleOperatorFactory.of(new LifeCycleTrackingOneInputStreamOperator()),
+ new PerRoundOperatorWrapper<>());
+ OperatorID operatorId = new OperatorID();
+
+ try (StreamTaskMailboxTestHarness<IterationRecord<Integer>> harness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
+ new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .setupOutputForSingletonOperatorChain(wrapperFactory, operatorId)
+ .build()) {
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(5, 1), 2));
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(6, 2), 3));
+
+ // Check the output
Review comment:
Checks ?
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputPerRoundWrapperOperator}. */
+public class OneInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
+
+ @Test
+ public void testProcessElementsAndEpochWatermarks() throws Exception {
+ StreamOperatorFactory<IterationRecord<Integer>> wrapperFactory =
+ new WrapperOperatorFactory<>(
+ SimpleOperatorFactory.of(new LifeCycleTrackingOneInputStreamOperator()),
+ new PerRoundOperatorWrapper<>());
+ OperatorID operatorId = new OperatorID();
+
+ try (StreamTaskMailboxTestHarness<IterationRecord<Integer>> harness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
+ new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .setupOutputForSingletonOperatorChain(wrapperFactory, operatorId)
+ .build()) {
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(5, 1), 2));
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(6, 2), 3));
+
+ // Check the output
+ assertEquals(
+ Arrays.asList(
+ new StreamRecord<>(IterationRecord.newRecord(5, 1), 2),
+ new StreamRecord<>(IterationRecord.newRecord(6, 2), 3)),
+ new ArrayList<>(harness.getOutput()));
+
+ // Check the other lifecycles.
Review comment:
Checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747283
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
##########
@@ -0,0 +1,482 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.AbstractWrapperOperator;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.proxy.state.ProxyStateSnapshotContext;
+import org.apache.flink.iteration.proxy.state.ProxyStreamOperatorStateContext;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactoryUtil;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.util.LatencyStats;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+/** The base class for all the per-round wrapper operators. */
+public abstract class AbstractPerRoundWrapperOperator<T, S extends StreamOperator<T>>
+ extends AbstractWrapperOperator<T>
+ implements StreamOperatorStateHandler.CheckpointedStreamOperator {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractPerRoundWrapperOperator.class);
+
+ /** The wrapped operators for each round. */
+ private final Map<Integer, S> wrappedOperators;
+
+ protected final LatencyStats latencyStats;
+
+ private transient StreamOperatorStateContext streamOperatorStateContext;
+
+ private transient StreamOperatorStateHandler stateHandler;
+
+ private transient InternalTimeServiceManager<?> timeServiceManager;
+
+ private transient KeySelector<?, ?> stateKeySelector1;
+
+ private transient KeySelector<?, ?> stateKeySelector2;
+
+ public AbstractPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<T>> parameters,
+ StreamOperatorFactory<T> operatorFactory) {
+ super(parameters, operatorFactory);
+
+ this.wrappedOperators = new HashMap<>();
+ this.latencyStats = initializeLatencyStats();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ protected S getWrappedOperator(int round) {
+ S wrappedOperator = wrappedOperators.get(round);
+ if (wrappedOperator != null) {
+ return wrappedOperator;
+ }
+
+ // We needs to clone the operator factory to also support SimpleOperatorFactory.
+ try {
+ StreamOperatorFactory<T> clonedOperatorFactory =
+ InstantiationUtil.clone(operatorFactory);
+ wrappedOperator =
+ (S)
+ StreamOperatorFactoryUtil.<T, S>createOperator(
+ clonedOperatorFactory,
+ (StreamTask) parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ proxyOutput,
+ parameters.getOperatorEventDispatcher())
+ .f0;
+ initializeStreamOperator(wrappedOperator, round);
+ wrappedOperators.put(round, wrappedOperator);
+ return wrappedOperator;
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+
+ return wrappedOperator;
+ }
+
+ @Override
+ public void onEpochWatermarkIncrement(int epochWatermark) throws IOException {
+ try {
+ // Deserts all the operators with round < epoch watermark. Notes that
+ // the onEpochWatermarkIncrement must be from 0 and increment by 1 each time.
+ if (wrappedOperators.containsKey(epochWatermark)) {
+ closeStreamOperator(wrappedOperators.get(epochWatermark), epochWatermark);
+ wrappedOperators.remove(epochWatermark);
+ }
+
+ super.onEpochWatermarkIncrement(epochWatermark);
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+ }
+
+ protected void processForEachWrappedOperator(
+ BiConsumerWithException<Integer, S, Exception> consumer) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ consumer.accept(entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Override
+ public void open() throws Exception {}
+
+ @Override
+ public void initializeState(StreamTaskStateInitializer streamTaskStateManager)
+ throws Exception {
+ final TypeSerializer<?> keySerializer =
+ streamConfig.getStateKeySerializer(containingTask.getUserCodeClassLoader());
+
+ streamOperatorStateContext =
+ streamTaskStateManager.streamOperatorStateContext(
+ getOperatorID(),
+ getClass().getSimpleName(),
+ parameters.getProcessingTimeService(),
+ this,
+ keySerializer,
+ containingTask.getCancelables(),
+ metrics,
+ streamConfig.getManagedMemoryFractionOperatorUseCaseOfSlot(
+ ManagedMemoryUseCase.STATE_BACKEND,
+ containingTask
+ .getEnvironment()
+ .getTaskManagerInfo()
+ .getConfiguration(),
+ containingTask.getUserCodeClassLoader()),
+ isUsingCustomRawKeyedState());
+
+ stateHandler =
+ new StreamOperatorStateHandler(
+ streamOperatorStateContext,
+ containingTask.getExecutionConfig(),
+ containingTask.getCancelables());
+ stateHandler.initializeOperatorState(this);
+ this.timeServiceManager = streamOperatorStateContext.internalTimerServiceManager();
+
+ stateKeySelector1 =
+ streamConfig.getStatePartitioner(0, containingTask.getUserCodeClassLoader());
+ stateKeySelector2 =
+ streamConfig.getStatePartitioner(1, containingTask.getUserCodeClassLoader());
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ // Do thing for now since we do not have states.
+ }
+
+ @Internal
+ protected boolean isUsingCustomRawKeyedState() {
+ return false;
+ }
+
+ @Override
+ public void finish() throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ closeStreamOperator(entry.getValue(), entry.getKey());
+ }
+ wrappedOperators.clear();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (stateHandler != null) {
+ stateHandler.dispose();
+ }
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().prepareSnapshotPreBarrier(checkpointId);
+ }
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory factory)
+ throws Exception {
+ return stateHandler.snapshotState(
+ this,
+ Optional.ofNullable(timeServiceManager),
+ streamConfig.getOperatorName(),
+ checkpointId,
+ timestamp,
+ checkpointOptions,
+ factory,
+ isUsingCustomRawKeyedState());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ if (StreamOperatorStateHandler.CheckpointedStreamOperator.class.isAssignableFrom(
+ entry.getValue().getClass())) {
+ ((StreamOperatorStateHandler.CheckpointedStreamOperator) entry.getValue())
+ .snapshotState(new ProxyStateSnapshotContext(context));
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement1(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector1);
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement2(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector2);
+ }
+
+ private <T> void setKeyContextElement(StreamRecord<T> record, KeySelector<T, ?> selector)
+ throws Exception {
+ if (selector != null) {
+ Object key = selector.getKey(record.getValue());
+ setCurrentKey(key);
+ }
+ }
+
+ @Override
+ public OperatorMetricGroup getMetricGroup() {
+ return metrics;
+ }
+
+ @Override
+ public OperatorID getOperatorID() {
+ return streamConfig.getOperatorID();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long l) throws Exception {
Review comment:
l -> checkpointId
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] gaoyunhaii commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739764735
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
##########
@@ -0,0 +1,482 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.AbstractWrapperOperator;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.proxy.state.ProxyStateSnapshotContext;
+import org.apache.flink.iteration.proxy.state.ProxyStreamOperatorStateContext;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactoryUtil;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.util.LatencyStats;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+/** The base class for all the per-round wrapper operators. */
+public abstract class AbstractPerRoundWrapperOperator<T, S extends StreamOperator<T>>
+ extends AbstractWrapperOperator<T>
+ implements StreamOperatorStateHandler.CheckpointedStreamOperator {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractPerRoundWrapperOperator.class);
+
+ /** The wrapped operators for each round. */
+ private final Map<Integer, S> wrappedOperators;
+
+ protected final LatencyStats latencyStats;
+
+ private transient StreamOperatorStateContext streamOperatorStateContext;
+
+ private transient StreamOperatorStateHandler stateHandler;
+
+ private transient InternalTimeServiceManager<?> timeServiceManager;
+
+ private transient KeySelector<?, ?> stateKeySelector1;
+
+ private transient KeySelector<?, ?> stateKeySelector2;
+
+ public AbstractPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<T>> parameters,
+ StreamOperatorFactory<T> operatorFactory) {
+ super(parameters, operatorFactory);
+
+ this.wrappedOperators = new HashMap<>();
+ this.latencyStats = initializeLatencyStats();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ protected S getWrappedOperator(int round) {
+ S wrappedOperator = wrappedOperators.get(round);
+ if (wrappedOperator != null) {
+ return wrappedOperator;
+ }
+
+ // We needs to clone the operator factory to also support SimpleOperatorFactory.
+ try {
+ StreamOperatorFactory<T> clonedOperatorFactory =
+ InstantiationUtil.clone(operatorFactory);
+ wrappedOperator =
+ (S)
+ StreamOperatorFactoryUtil.<T, S>createOperator(
+ clonedOperatorFactory,
+ (StreamTask) parameters.getContainingTask(),
+ parameters.getStreamConfig(),
+ proxyOutput,
+ parameters.getOperatorEventDispatcher())
+ .f0;
+ initializeStreamOperator(wrappedOperator, round);
+ wrappedOperators.put(round, wrappedOperator);
+ return wrappedOperator;
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+
+ return wrappedOperator;
+ }
+
+ @Override
+ public void onEpochWatermarkIncrement(int epochWatermark) throws IOException {
+ try {
+ // Deserts all the operators with round < epoch watermark. Notes that
+ // the onEpochWatermarkIncrement must be from 0 and increment by 1 each time.
+ if (wrappedOperators.containsKey(epochWatermark)) {
+ closeStreamOperator(wrappedOperators.get(epochWatermark), epochWatermark);
+ wrappedOperators.remove(epochWatermark);
+ }
+
+ super.onEpochWatermarkIncrement(epochWatermark);
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+ }
+
+ protected void processForEachWrappedOperator(
+ BiConsumerWithException<Integer, S, Exception> consumer) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ consumer.accept(entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Override
+ public void open() throws Exception {}
+
+ @Override
+ public void initializeState(StreamTaskStateInitializer streamTaskStateManager)
+ throws Exception {
+ final TypeSerializer<?> keySerializer =
+ streamConfig.getStateKeySerializer(containingTask.getUserCodeClassLoader());
+
+ streamOperatorStateContext =
+ streamTaskStateManager.streamOperatorStateContext(
+ getOperatorID(),
+ getClass().getSimpleName(),
+ parameters.getProcessingTimeService(),
+ this,
+ keySerializer,
+ containingTask.getCancelables(),
+ metrics,
+ streamConfig.getManagedMemoryFractionOperatorUseCaseOfSlot(
+ ManagedMemoryUseCase.STATE_BACKEND,
+ containingTask
+ .getEnvironment()
+ .getTaskManagerInfo()
+ .getConfiguration(),
+ containingTask.getUserCodeClassLoader()),
+ isUsingCustomRawKeyedState());
+
+ stateHandler =
+ new StreamOperatorStateHandler(
+ streamOperatorStateContext,
+ containingTask.getExecutionConfig(),
+ containingTask.getCancelables());
+ stateHandler.initializeOperatorState(this);
+ this.timeServiceManager = streamOperatorStateContext.internalTimerServiceManager();
+
+ stateKeySelector1 =
+ streamConfig.getStatePartitioner(0, containingTask.getUserCodeClassLoader());
+ stateKeySelector2 =
+ streamConfig.getStatePartitioner(1, containingTask.getUserCodeClassLoader());
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ // Do thing for now since we do not have states.
+ }
+
+ @Internal
+ protected boolean isUsingCustomRawKeyedState() {
+ return false;
+ }
+
+ @Override
+ public void finish() throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ closeStreamOperator(entry.getValue(), entry.getKey());
+ }
+ wrappedOperators.clear();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (stateHandler != null) {
+ stateHandler.dispose();
+ }
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().prepareSnapshotPreBarrier(checkpointId);
+ }
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory factory)
+ throws Exception {
+ return stateHandler.snapshotState(
+ this,
+ Optional.ofNullable(timeServiceManager),
+ streamConfig.getOperatorName(),
+ checkpointId,
+ timestamp,
+ checkpointOptions,
+ factory,
+ isUsingCustomRawKeyedState());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ if (StreamOperatorStateHandler.CheckpointedStreamOperator.class.isAssignableFrom(
+ entry.getValue().getClass())) {
+ ((StreamOperatorStateHandler.CheckpointedStreamOperator) entry.getValue())
+ .snapshotState(new ProxyStateSnapshotContext(context));
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement1(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector1);
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement2(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector2);
+ }
+
+ private <T> void setKeyContextElement(StreamRecord<T> record, KeySelector<T, ?> selector)
+ throws Exception {
+ if (selector != null) {
+ Object key = selector.getKey(record.getValue());
+ setCurrentKey(key);
+ }
+ }
+
+ @Override
+ public OperatorMetricGroup getMetricGroup() {
+ return metrics;
+ }
+
+ @Override
+ public OperatorID getOperatorID() {
+ return streamConfig.getOperatorID();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long l) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().notifyCheckpointComplete(l);
+ }
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ for (Map.Entry<Integer, S> entry : wrappedOperators.entrySet()) {
+ entry.getValue().notifyCheckpointAborted(checkpointId);
+ }
+ }
+
+ @Override
+ public void setCurrentKey(Object key) {
+ stateHandler.setCurrentKey(key);
+ }
+
+ @Override
+ public Object getCurrentKey() {
+ if (stateHandler == null) {
+ return null;
+ }
+
+ return stateHandler.getKeyedStateStore().orElse(null);
+ }
+
+ protected void reportOrForwardLatencyMarker(LatencyMarker marker) {
+ // all operators are tracking latencies
+ this.latencyStats.reportLatency(marker);
+
+ // everything except sinks forwards latency markers
+ this.output.emitLatencyMarker(marker);
+ }
+
+ private LatencyStats initializeLatencyStats() {
Review comment:
Sorry I have a double check with the currently implementation, it seems only the wrapper operators process the latency mark, thus perhaps we could first keep it as now?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747095
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.operator.allround.OneInputAllRoundWrapperOperator;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractInput;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2;
+import org.apache.flink.streaming.api.operators.Input;
+import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.MultipleInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputAllRoundWrapperOperator}. */
+public class MultipleInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
+
+ @Test
+ public void testProcessElementsAndEpochWatermarks() throws Exception {
+ StreamOperatorFactory<IterationRecord<Integer>> wrapperFactory =
+ new WrapperOperatorFactory<>(
+ new LifeCycleTrackingMultiInputStreamOperatorFactory(),
+ new PerRoundOperatorWrapper<>());
+ OperatorID operatorId = new OperatorID();
+
+ try (StreamTaskMailboxTestHarness<IterationRecord<Integer>> harness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ MultipleInputStreamTask::new,
+ new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .setupOutputForSingletonOperatorChain(wrapperFactory, operatorId)
+ .build()) {
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(5, 1), 2), 0);
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(6, 2), 3), 2);
+
+ // Check the output
Review comment:
Checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739747243
##########
File path: flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorUtils;
+import org.apache.flink.iteration.operator.WrapperOperatorFactory;
+import org.apache.flink.iteration.operator.allround.LifeCycle;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the {@link OneInputPerRoundWrapperOperator}. */
+public class OneInputPerRoundWrapperOperatorTest {
+
+ private static List<LifeCycle> lifeCycles = new ArrayList<>();
+
+ @Test
+ public void testProcessElementsAndEpochWatermarks() throws Exception {
+ StreamOperatorFactory<IterationRecord<Integer>> wrapperFactory =
+ new WrapperOperatorFactory<>(
+ SimpleOperatorFactory.of(new LifeCycleTrackingOneInputStreamOperator()),
+ new PerRoundOperatorWrapper<>());
+ OperatorID operatorId = new OperatorID();
+
+ try (StreamTaskMailboxTestHarness<IterationRecord<Integer>> harness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
+ new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .addInput(new IterationRecordTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO))
+ .setupOutputForSingletonOperatorChain(wrapperFactory, operatorId)
+ .build()) {
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(5, 1), 2));
+ harness.processElement(new StreamRecord<>(IterationRecord.newRecord(6, 2), 3));
+
+ // Check the output
+ assertEquals(
+ Arrays.asList(
+ new StreamRecord<>(IterationRecord.newRecord(5, 1), 2),
+ new StreamRecord<>(IterationRecord.newRecord(6, 2), 3)),
+ new ArrayList<>(harness.getOutput()));
+
+ // Check the other lifecycles.
+ harness.getStreamTask()
+ .triggerCheckpointOnBarrier(
+ new CheckpointMetaData(5, 2),
+ CheckpointOptions.alignedNoTimeout(
+ CheckpointType.CHECKPOINT,
+ CheckpointStorageLocationReference.getDefault()),
+ new CheckpointMetricsBuilder()
+ .setAlignmentDurationNanos(0)
+ .setBytesProcessedDuringAlignment(0));
+ harness.processAll();
+
+ harness.getStreamTask().notifyCheckpointCompleteAsync(5);
+ harness.processAll();
+
+ harness.getStreamTask().notifyCheckpointAbortAsync(6, 5);
+ harness.processAll();
+
+ harness.getOutput().clear();
+ harness.processElement(
+ new StreamRecord<>(IterationRecord.newEpochWatermark(1, "only-one")));
+ harness.processElement(
+ new StreamRecord<>(IterationRecord.newEpochWatermark(2, "only-one")));
+
+ // Check the output
Review comment:
Checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r725419380
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/operator/perrond/MultipleInputPerRoundWrapperOperator.java
##########
@@ -0,0 +1,131 @@
+/*
Review comment:
There's a typo in the name of the folder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] guoweiM commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739746923
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/proxy/state/ProxyOperatorStateBackend.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.iteration.proxy.state;
+
+import org.apache.flink.api.common.state.BroadcastState;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.SnapshotResult;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.RunnableFuture;
+
+/** Proxy {@link OperatorStateBackend} for the wrapped Operator. */
+public class ProxyOperatorStateBackend implements OperatorStateBackend {
+
+ private final OperatorStateBackend wrappedBackend;
+
+ private final String stateNamePrefix;
+
+ public ProxyOperatorStateBackend(OperatorStateBackend wrappedBackend, String stateNamePrefix) {
+ this.wrappedBackend = wrappedBackend;
+ this.stateNamePrefix = stateNamePrefix;
+ }
+
+ @Override
+ public <K, V> BroadcastState<K, V> getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)
+ throws Exception {
+ MapStateDescriptor<K, V> newDescriptor =
+ new MapStateDescriptor<>(
+ stateNamePrefix + stateDescriptor.getName(),
Review comment:
I think it might be more trackable that we make a `stateNamePrefix + stateDescriptor.getName()` common method. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] gaoyunhaii commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739765096
##########
File path: flink-ml-iteration/pom.xml
##########
@@ -61,6 +61,14 @@ under the License.
<scope>provided</scope>
</dependency>
+ <!-- We have special treatment with the rocksdb state. -->
Review comment:
I think we might not since in the main code we have references to rocksdb related classes. With provided, the dependency would also not be included into the jar.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] gaoyunhaii commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739765187
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperator.java
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.streaming.api.graph.StreamEdge;
+import org.apache.flink.streaming.api.operators.Input;
+import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Per-round wrapper for the multiple-inputs operator. */
+public class MultipleInputPerRoundWrapperOperator<OUT>
+ extends AbstractPerRoundWrapperOperator<OUT, MultipleInputStreamOperator<OUT>>
+ implements MultipleInputStreamOperator<IterationRecord<OUT>> {
+
+ public MultipleInputPerRoundWrapperOperator(
+ StreamOperatorParameters<IterationRecord<OUT>> parameters,
+ StreamOperatorFactory<OUT> operatorFactory) {
+ super(parameters, operatorFactory);
+ }
+
+ private <IN> void processElement(
+ int inputIndex,
+ Input<IN> input,
+ StreamRecord<IN> reusedInput,
+ StreamRecord<IterationRecord<IN>> element)
+ throws Exception {
+ switch (element.getValue().getType()) {
+ case RECORD:
+ reusedInput.replace(element.getValue().getValue(), element.getTimestamp());
+ setIterationContextRound(element.getValue().getEpoch());
+ input.processElement(reusedInput);
+ clearIterationContextRound();
+ break;
+ case EPOCH_WATERMARK:
+ onEpochWatermarkEvent(inputIndex, element.getValue());
+ break;
+ default:
+ throw new FlinkRuntimeException("Not supported iteration record type: " + element);
+ }
+ }
+
+ @Override
+ @SuppressWarnings({"rawtypes"})
+ public List<Input> getInputs() {
+ List<Input> proxyInputs = new ArrayList<>();
+
+ // Determine how much inputs we have
+ List<StreamEdge> inEdges =
+ streamConfig.getInPhysicalEdges(containingTask.getUserCodeClassLoader());
+ int numberOfInputs =
+ inEdges.stream().map(StreamEdge::getTypeNumber).collect(Collectors.toSet()).size();
+
+ for (int i = 0; i < numberOfInputs; ++i) {
+ // TODO: Note that here we relies on the assumption that the
Review comment:
To remove the assumption we would need to keep an explicit maps for the mapping. We might consider the issue later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] gaoyunhaii commented on a change in pull request #14: [FLINK-8][iteration] Add per-round operator wrappers
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #14:
URL: https://github.com/apache/flink-ml/pull/14#discussion_r739764795
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/PerRoundOperatorWrapper.java
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.iteration.operator.perround;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.iteration.IterationRecord;
+import org.apache.flink.iteration.operator.OperatorWrapper;
+import org.apache.flink.iteration.proxy.ProxyKeySelector;
+import org.apache.flink.iteration.proxy.ProxyStreamPartitioner;
+import org.apache.flink.iteration.typeinfo.IterationRecordTypeInfo;
+import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
+import org.apache.flink.util.OutputTag;
+
+/** The operator wrapper implementation for per-round wrappers. */
+public class PerRoundOperatorWrapper<T> implements OperatorWrapper<T, IterationRecord<T>> {
+
+ @Override
+ public StreamOperator<IterationRecord<T>> wrap(
Review comment:
We supported the empty iteration body in the previous PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org