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/26 07:50:53 UTC
[GitHub] [flink-ml] gaoyunhaii opened a new pull request #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
gaoyunhaii opened a new pull request #6:
URL: https://github.com/apache/flink-ml/pull/6
Iteration requires to broadcast the epoch-watermark inside iteration no matter what partitioner is used.
Currently the functionality is implementation via reflection. As a whole, the output of the an operator might be
1. ChainingOutput: if the following operators is chained with this one. In this case, we could direct output the event, which id equivalent to broadcast.
2. RecordWriterOutput: the records are written to an external task. The output wraps an `RecordWriter`, which provides the functionality of broadcast to support watermark / latencymark, and we need to proxy broadcasting to this object via reflection.
3. BroadcastingOutput: The operator itself have multiple downstream operators / tasks. `BroadcastingOutput` has a list of internal outputs, which must be 1 or 2.
We should also maintain the metric of `numOutputRecords` by counting the broadcasted event.
--
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 pull request #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#issuecomment-950341636
Very thanks @guoweiM for the review! I updated the PR to address the inline comments~
I also agree with we should push the broadcast functionality back to the runtime. Currently there is an ongoing related discussion of broadcast events, we would also follow the discussion~
--
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] lindong28 commented on pull request #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#issuecomment-927299201
Thanks for the PR @gaoyunhaii. It is not clear why FLIP-176 needs this `BroadcastOutput`. Can you explain it? And is there any existing end-to-end test that shows how to support an iteration job using this `BroadcastOutput`?
--
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] becketqin commented on a change in pull request #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
becketqin commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r719874290
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/broadcast/OutputReflectionContext.java
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.ml.iteration.utils.ReflectionUtils;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.io.RecordWriterOutput;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.lang.reflect.Field;
+
+/** The reflection utilities to parse the output and create the broadcast output */
+public class OutputReflectionContext {
Review comment:
Should this class be a util class with everything static? Or at least a singleton class?
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/broadcast/BroadcastOutputFactory.java
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Factor that creates the corresponding {@link BroadcastOutput} from the {@link Output}. */
+public class BroadcastOutputFactory {
+
+ /**
+ * Creates the wrapper broadcast ooutput from {@code output}.
+ *
+ * @param output the original output.
+ * @param numRecordsOut the counter for the number of output record.
+ * @return the wrapped broadcast output.
+ */
+ public static <OUT> BroadcastOutput<OUT> createBroadcastOutput(
+ Output<StreamRecord<OUT>> output, Counter numRecordsOut) {
+
+ OutputReflectionContext outputReflectionContext = new OutputReflectionContext();
+
+ List<BroadcastOutput<OUT>> internalOutputs = new ArrayList<>();
+ if (outputReflectionContext.isBroadcastingOutput(output)) {
Review comment:
Is it possible to have nested `BroadcastingOutput`?
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/broadcast/ChainingBroadcastOutput.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
+public class ChainingBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
Review comment:
Is this BroadcastOutput only emitting to the downstream operator or to side output, but not both?
--
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 #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r720815462
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/broadcast/BroadcastOutputFactory.java
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Factor that creates the corresponding {@link BroadcastOutput} from the {@link Output}. */
+public class BroadcastOutputFactory {
+
+ /**
+ * Creates the wrapper broadcast ooutput from {@code output}.
+ *
+ * @param output the original output.
+ * @param numRecordsOut the counter for the number of output record.
+ * @return the wrapped broadcast output.
+ */
+ public static <OUT> BroadcastOutput<OUT> createBroadcastOutput(
+ Output<StreamRecord<OUT>> output, Counter numRecordsOut) {
+
+ OutputReflectionContext outputReflectionContext = new OutputReflectionContext();
+
+ List<BroadcastOutput<OUT>> internalOutputs = new ArrayList<>();
+ if (outputReflectionContext.isBroadcastingOutput(output)) {
Review comment:
No, it is not possible with the current implementation.
--
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 #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r720815419
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/broadcast/OutputReflectionContext.java
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.ml.iteration.utils.ReflectionUtils;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.io.RecordWriterOutput;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.lang.reflect.Field;
+
+/** The reflection utilities to parse the output and create the broadcast output */
+public class OutputReflectionContext {
Review comment:
I'm a bit hesitate here in that I'm not sure java reflection library is thread safe. I think it should not cause to much performance penalty to keep one context for each operator.
--
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 #6: [FLINK-24644][iteration] Introduce broadcast output to support broadcast events to all the downstream operators
Posted by GitBox <gi...@apache.org>.
gaoyunhaii closed pull request #6:
URL: https://github.com/apache/flink-ml/pull/6
--
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 #6: [FLINK-24644][iteration] Introduce broadcast output to support broadcast events to all the downstream operators
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r737313399
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
Review comment:
😅 thanks for your explanation.
--
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 #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r735131924
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
Review comment:
Unfortunately the `ChainingOutput` is protected and is not accessible here.
--
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 #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r732456097
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
+public class ChainingBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
+ private final Output<StreamRecord<OUT>> rawOutput;
+ private final OutputTag outputTag;
+
+ ChainingBroadcastOutput(Output<StreamRecord<OUT>> rawOutput, OutputTag outputTag) {
+ this.rawOutput = rawOutput;
+ this.outputTag = outputTag;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void broadcastEmit(StreamRecord<OUT> record) {
Review comment:
If I understand correctly, broadcast means that no matter what the tag is, the data should be broadcast down. If this is the case, I understand that rawOutput.collect should be fine here?
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/BroadcastOutputFactory.java
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Factor that creates the corresponding {@link BroadcastOutput} from the {@link Output}. */
Review comment:
Factory
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
Review comment:
Here I suggest that you can refer to specific classes, such as {@linkto ChainingOutput}
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/BroadcastOutputFactory.java
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Factor that creates the corresponding {@link BroadcastOutput} from the {@link Output}. */
+public class BroadcastOutputFactory {
+
+ /**
+ * Creates the wrapper broadcast ooutput from {@code output}.
Review comment:
ooutput -> output
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/OutputReflectionContext.java
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.io.RecordWriterOutput;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.lang.reflect.Field;
+
+/** The reflection utilities to parse the output and create the broadcast output. */
+public class OutputReflectionContext {
+ private final Class<?> broadcastingOutputClass;
+ private final Field broadcastingOutputsField;
+
+ private final Class<?> chainingOutputClass;
+ private final Field chainingOutputTagField;
+
+ private final Field recordWriterField;
+ private final Field recordWriterSerializationDelegateField;
+ private final Field serializationDelegateSerializerField;
+
+ public OutputReflectionContext() {
+ try {
+ this.broadcastingOutputClass =
+ Class.forName(
+ "org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector");
+ this.broadcastingOutputsField =
+ ReflectionUtils.getClassField(broadcastingOutputClass, "outputs");
+
+ this.chainingOutputClass =
+ Class.forName("org.apache.flink.streaming.runtime.tasks.ChainingOutput");
+ this.chainingOutputTagField =
+ ReflectionUtils.getClassField(chainingOutputClass, "outputTag");
+
+ this.recordWriterField =
+ ReflectionUtils.getClassField(RecordWriterOutput.class, "recordWriter");
+ this.recordWriterSerializationDelegateField =
+ ReflectionUtils.getClassField(
+ RecordWriterOutput.class, "serializationDelegate");
+ this.serializationDelegateSerializerField =
+ ReflectionUtils.getClassField(SerializationDelegate.class, "serializer");
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to initialize the OutputReflectionContext", e);
+ }
+ }
+
+ public boolean isBroadcastingOutput(Output<?> bareOutput) {
Review comment:
bareOutput ---> rawOutput
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/CountingBroadcastOutput.java
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * The intermediate broadcast output that wrappers a list of internal outputs. It will broadcast the
+ * records to all the internal outputs and increment the counter.
+ */
+public class CountingBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
+
+ private final Counter numRecordsOut;
+ private final List<BroadcastOutput<OUT>> internalOutputs;
+
+ public CountingBroadcastOutput(
+ Counter numRecordsOut, List<BroadcastOutput<OUT>> internalOutputs) {
+ this.numRecordsOut = numRecordsOut;
+ this.internalOutputs = internalOutputs;
+ }
+
+ @Override
+ public void broadcastEmit(StreamRecord<OUT> record) throws IOException {
+ numRecordsOut.inc();
Review comment:
Why does the broadcasted record only increment once?
--
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 #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r730742537
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/ml/iteration/broadcast/ChainingBroadcastOutput.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
+public class ChainingBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
Review comment:
Yes, a chained output must either has output tag or not.
--
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 pull request #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#issuecomment-945594305
Very thanks @lindong28 and @becketqin for the review!
> Thanks for the PR @gaoyunhaii. It is not clear why FLIP-176 needs this BroadcastOutput. Can you explain it? And is there > any existing end-to-end test that shows how to support an iteration job using this BroadcastOutput?
As @becketqin has pointed out, this utility is required mainly due to that we need to broadcast the epoch event no matter what the shuffler partitioner is used. The usage of this utility is available in the unit tests and in the following PR~
> If so is it the case, is this broadcast output only needed in case of the shuffle? Is my understanding correct?
Flink has two kinds of outputs for an operator:
1. ChainedOutput: the output to another operator in the same operator chain.
2. RecordWriterOutput: the output to another operator in another task.
If one operator has multiple outputs, they will be wrapped in a `BroadcastOutput`.
Thus logically we only need to deal with the `RecordWriterOutput`. However, for `ChainedOutput` we still need to deal with the `OutputTag`: if the output is defined with an output tag, we need to also emit with this output tag for the events, otherwise these events would be deserted.
Thus we would need to deal with both types of outputs.
--
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 #6: [FLINK-24644][iteration] Introduce broadcast output to support broadcast events to all the downstream operators
Posted by GitBox <gi...@apache.org>.
guoweiM commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r737313399
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
Review comment:
😅 thanks for your expalination.
--
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 #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r735132139
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/CountingBroadcastOutput.java
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * The intermediate broadcast output that wrappers a list of internal outputs. It will broadcast the
+ * records to all the internal outputs and increment the counter.
+ */
+public class CountingBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
+
+ private final Counter numRecordsOut;
+ private final List<BroadcastOutput<OUT>> internalOutputs;
+
+ public CountingBroadcastOutput(
+ Counter numRecordsOut, List<BroadcastOutput<OUT>> internalOutputs) {
+ this.numRecordsOut = numRecordsOut;
+ this.internalOutputs = internalOutputs;
+ }
+
+ @Override
+ public void broadcastEmit(StreamRecord<OUT> record) throws IOException {
+ numRecordsOut.inc();
Review comment:
Since from the view of the this operator it indeed only emit "one record", which is also consistency with the model for the ordinary broadcast partitioner~
--
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 #6: [FLINK-2][iteration] Add broadcast output to broadcast events to all the downstream tasks
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #6:
URL: https://github.com/apache/flink-ml/pull/6#discussion_r735131743
##########
File path: flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
+public class ChainingBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
+ private final Output<StreamRecord<OUT>> rawOutput;
+ private final OutputTag outputTag;
+
+ ChainingBroadcastOutput(Output<StreamRecord<OUT>> rawOutput, OutputTag outputTag) {
+ this.rawOutput = rawOutput;
+ this.outputTag = outputTag;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void broadcastEmit(StreamRecord<OUT> record) {
Review comment:
This is due to we have to deal with both cases of with OutputTag and without OutputTag. For the broadcast cases, we required the record must be send no matter the OutputTag is set. If it is indeed set, `rawOutput.collect` would ignore the record.
--
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