You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "GreatEugenius (via GitHub)" <gi...@apache.org> on 2025/06/26 02:57:30 UTC
[PR] [Feature][runtime] Introducing the FlinkRunnerContext component to support Python workflow execution [flink-agents]
GreatEugenius opened a new pull request, #43:
URL: https://github.com/apache/flink-agents/pull/43
<!--
* Thank you very much for contributing to Flink Agents.
* Please add the relevant components in the PR title. E.g., [api], [runtime], [java], [python], [hotfix], etc.
-->
<!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this. -->
Linked issue: #27
### Purpose of change
<!-- What is the purpose of this change? -->
Introducing the FlinkRunnerContext component to support Python workflow execution.
### Tests
<!-- How is this change verified? -->
### API
<!-- Does this change touches any public APIs? -->
### Documentation
<!-- Should this change be covered by the user documentation?-->
--
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
Re: [PR] [Feature][runtime] Introducing the FlinkRunnerContext component to support Python workflow execution [flink-agents]
Posted by "GreatEugenius (via GitHub)" <gi...@apache.org>.
GreatEugenius commented on code in PR #43:
URL: https://github.com/apache/flink-agents/pull/43#discussion_r2174301665
##########
python/flink_agents/plan/function.py:
##########
@@ -141,3 +141,9 @@ def __call__(self, *args: Tuple[Any, ...], **kwargs: Dict[str, Any]) -> Any:
def check_signature(self, *args: Tuple[Any, ...]) -> None:
"""Check function signature is legal or not."""
+
+
+def call_python_function(module: str, qualname: str, func_args: Tuple[Any, ...]) -> Any:
+ """Used to call a Python function in the Pemja environment."""
+ func = PythonFunction(module=module, qualname=qualname)
Review Comment:
Thanks for the feedback! You're absolutely right—repeatedly creating PythonFunction instances could lead to unnecessary overhead. I've created a follow-up issue to track this optimization: [Issue #46](https://github.com/apache/flink-agents/issues/46).
--
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
Re: [PR] [Feature][runtime] Introducing the FlinkRunnerContext component to support Python workflow execution [flink-agents]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #43:
URL: https://github.com/apache/flink-agents/pull/43#discussion_r2168047055
##########
runtime/src/main/java/org/apache/flink/agents/runtime/PythonRunner.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.agents.runtime;
+
+import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.runtime.context.PythonRunnerContext;
+import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
+import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+import pemja.core.PythonInterpreter;
+
+import java.util.List;
+
+/** Execute the corresponding Python action in the workflow. */
+public class PythonRunner {
+
+ private static final String IMPORT_FLINK_RUNNER_CONTEXT =
+ "from flink_agents.runtime import flink_runner_context";
+ private static final String CREATE_FLINK_RUNNER_CONTEXT =
+ "flink_runner_context.create_flink_runner_context";
+ private static final String CONVERT_TO_PYTHON_OBJECT =
+ "flink_runner_context.convert_to_python_object";
+
+ private PythonInterpreter interpreter;
+ private PythonRunnerContext runnerContext;
+
+ public PythonRunner(PythonEnvironmentManager environmentManager) throws Exception {
+ environmentManager.open();
+ EmbeddedPythonEnvironment env =
+ (EmbeddedPythonEnvironment) environmentManager.createEnvironment();
+
+ interpreter = env.getInterpreter();
+ interpreter.exec(IMPORT_FLINK_RUNNER_CONTEXT);
+ }
+
+ public List executePythonFunction(
+ PythonFunction pythonFunction, PythonEventMessage<?> inputMessage) {
+ String modulePath = pythonFunction.getModule();
+ int moduleLastDotIndex = modulePath.lastIndexOf(".");
+ if (moduleLastDotIndex == -1) {
+ throw new IllegalArgumentException(
+ "Invalid Python module path: " + pythonFunction.getModule());
+ }
+
+ String packagePath = modulePath.substring(0, moduleLastDotIndex);
+ String className = modulePath.substring(moduleLastDotIndex + 1);
Review Comment:
This seems not correct. Why would the module path includes the class name?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/PythonRunner.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.agents.runtime;
+
+import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.runtime.context.PythonRunnerContext;
+import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
+import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+import pemja.core.PythonInterpreter;
+
+import java.util.List;
+
+/** Execute the corresponding Python action in the workflow. */
+public class PythonRunner {
+
+ private static final String IMPORT_FLINK_RUNNER_CONTEXT =
+ "from flink_agents.runtime import flink_runner_context";
+ private static final String CREATE_FLINK_RUNNER_CONTEXT =
+ "flink_runner_context.create_flink_runner_context";
+ private static final String CONVERT_TO_PYTHON_OBJECT =
+ "flink_runner_context.convert_to_python_object";
+
+ private PythonInterpreter interpreter;
+ private PythonRunnerContext runnerContext;
+
+ public PythonRunner(PythonEnvironmentManager environmentManager) throws Exception {
+ environmentManager.open();
+ EmbeddedPythonEnvironment env =
+ (EmbeddedPythonEnvironment) environmentManager.createEnvironment();
+
+ interpreter = env.getInterpreter();
+ interpreter.exec(IMPORT_FLINK_RUNNER_CONTEXT);
+ }
+
+ public List executePythonFunction(
Review Comment:
It would be better to implement `PythonFunction.call()`.
1. set the interpreter to the `PythonFunction`
2. call `PythonFunction.call()` and pass in the java arguments
3. convert java arguments into python objects and call the python version `PythonFunction.call()` in the java `PythonFunction.call()`
##########
runtime/src/main/java/org/apache/flink/agents/runtime/PythonRunner.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.agents.runtime;
+
+import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.runtime.context.PythonRunnerContext;
+import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
+import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+import pemja.core.PythonInterpreter;
+
+import java.util.List;
+
+/** Execute the corresponding Python action in the workflow. */
+public class PythonRunner {
+
+ private static final String IMPORT_FLINK_RUNNER_CONTEXT =
+ "from flink_agents.runtime import flink_runner_context";
+ private static final String CREATE_FLINK_RUNNER_CONTEXT =
+ "flink_runner_context.create_flink_runner_context";
+ private static final String CONVERT_TO_PYTHON_OBJECT =
+ "flink_runner_context.convert_to_python_object";
+
+ private PythonInterpreter interpreter;
+ private PythonRunnerContext runnerContext;
Review Comment:
This is never initialized.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/PythonRunnerContext.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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.agents.runtime.context;
+
+import org.apache.flink.agents.runtime.message.EventMessage;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This class is used to manage the execution context when interacting with Python functions,
+ * including temporarily saving the key of the current Event and collecting new events generated
+ * during the execution of actions.
+ */
+@NotThreadSafe
+public class PythonRunnerContext {
+ private Object key;
+ private List<EventMessage<?, ?>> events;
+
+ public PythonRunnerContext() {
+ this(null);
+ }
+
+ public PythonRunnerContext(Object key) {
+ this.key = key;
+ this.events = new ArrayList<>();
+ }
+
+ public void setKey(Object key) {
+ this.key = key;
+ }
+
+ public void sendEvent(String type, byte[] event) {
+ if (key == null) {
+ throw new IllegalStateException("Key is not set.");
+ }
+ this.events.add(new PythonEventMessage<>(key, event, type));
+ }
+
+ public List<EventMessage<?, ?>> drainEvents() {
+ List<EventMessage<?, ?>> list = new ArrayList<>(this.events);
+ clearAllEvents();
+ return list;
+ }
+
+ public void clearAllEvents() {
+ this.events.clear();
+ }
Review Comment:
Why separating into two methods?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/PythonRunner.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.agents.runtime;
+
+import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.runtime.context.PythonRunnerContext;
+import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
+import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+import pemja.core.PythonInterpreter;
+
+import java.util.List;
+
+/** Execute the corresponding Python action in the workflow. */
+public class PythonRunner {
Review Comment:
```suggestion
public class PythonActionExecutor {
```
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/PythonRunnerContext.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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.agents.runtime.context;
+
+import org.apache.flink.agents.runtime.message.EventMessage;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This class is used to manage the execution context when interacting with Python functions,
+ * including temporarily saving the key of the current Event and collecting new events generated
+ * during the execution of actions.
+ */
+@NotThreadSafe
+public class PythonRunnerContext {
+ private Object key;
+ private List<EventMessage<?, ?>> events;
+
+ public PythonRunnerContext() {
+ this(null);
+ }
+
+ public PythonRunnerContext(Object key) {
+ this.key = key;
+ this.events = new ArrayList<>();
+ }
+
+ public void setKey(Object key) {
+ this.key = key;
+ }
+
+ public void sendEvent(String type, byte[] event) {
+ if (key == null) {
+ throw new IllegalStateException("Key is not set.");
+ }
+ this.events.add(new PythonEventMessage<>(key, event, type));
Review Comment:
And why do we need to wrap the events into messages at all?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/PythonRunner.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.agents.runtime;
+
+import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.runtime.context.PythonRunnerContext;
+import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
+import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+import pemja.core.PythonInterpreter;
+
+import java.util.List;
+
+/** Execute the corresponding Python action in the workflow. */
+public class PythonRunner {
+
+ private static final String IMPORT_FLINK_RUNNER_CONTEXT =
+ "from flink_agents.runtime import flink_runner_context";
+ private static final String CREATE_FLINK_RUNNER_CONTEXT =
+ "flink_runner_context.create_flink_runner_context";
+ private static final String CONVERT_TO_PYTHON_OBJECT =
+ "flink_runner_context.convert_to_python_object";
+
+ private PythonInterpreter interpreter;
Review Comment:
Should this be `final`?
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -0,0 +1,68 @@
+################################################################################
+# 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.
+#################################################################################
+from typing import Any
+
+import cloudpickle
+from typing_extensions import override
+
+from flink_agents.api.event import Event, OutputEvent
+from flink_agents.api.runner_context import RunnerContext
+
+
+def create_flink_runner_context(j_runner_context: Any) -> "FlinkRunnerContext":
+ """Used to create a FlinkRunnerContext Python object in Pemja environment."""
+ return FlinkRunnerContext(j_runner_context)
+
+
+def convert_to_python_object(bytesObject: bytes) -> Any:
+ """Used for deserializing Python objects."""
+ return cloudpickle.loads(bytesObject)
Review Comment:
Why is this in the `flink_runner_context` module?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/PythonRunnerContext.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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.agents.runtime.context;
+
+import org.apache.flink.agents.runtime.message.EventMessage;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This class is used to manage the execution context when interacting with Python functions,
+ * including temporarily saving the key of the current Event and collecting new events generated
+ * during the execution of actions.
+ */
+@NotThreadSafe
+public class PythonRunnerContext {
+ private Object key;
+ private List<EventMessage<?, ?>> events;
+
+ public PythonRunnerContext() {
+ this(null);
+ }
+
+ public PythonRunnerContext(Object key) {
+ this.key = key;
+ this.events = new ArrayList<>();
+ }
+
+ public void setKey(Object key) {
+ this.key = key;
+ }
+
+ public void sendEvent(String type, byte[] event) {
+ if (key == null) {
+ throw new IllegalStateException("Key is not set.");
+ }
+ this.events.add(new PythonEventMessage<>(key, event, type));
Review Comment:
Why not wrapping all events into one message?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/message/PythonEventMessage.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.agents.runtime.message;
+
+import java.util.Arrays;
+
+/**
+ * PythonEventMessage is a {@link EventMessage} that carries a Python event in byte array format.
+ *
+ * @param <K> The type of the key.
+ */
+public class PythonEventMessage<K> extends EventMessage<K, byte[]> {
Review Comment:
What is the purpose of this wrap?
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -0,0 +1,68 @@
+################################################################################
+# 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.
+#################################################################################
+from typing import Any
+
+import cloudpickle
+from typing_extensions import override
+
+from flink_agents.api.event import Event, OutputEvent
+from flink_agents.api.runner_context import RunnerContext
+
+
+def create_flink_runner_context(j_runner_context: Any) -> "FlinkRunnerContext":
+ """Used to create a FlinkRunnerContext Python object in Pemja environment."""
+ return FlinkRunnerContext(j_runner_context)
+
+
+def convert_to_python_object(bytesObject: bytes) -> Any:
+ """Used for deserializing Python objects."""
+ return cloudpickle.loads(bytesObject)
+
+
+class FlinkRunnerContext(RunnerContext):
+ """Providing context for workflow execution in Flink Environment.
+
+ This context allows access to event handling.
+ """
+
+ def __init__(self, j_runner_context: Any) -> None:
+ """Initialize a flink runner context with the given java runner context.
+
+ Parameters
+ ----------
+ j_runner_context : Any
+ Java runner context used to synchronize data between Python and Java.
+ """
+ self._j_runner_context = j_runner_context
+
+ @override
+ def send_event(self, event: Event) -> None:
+ """Send an event to the workflow for processing.
+
+ Parameters
+ ----------
+ event : Event
+ The event to be processed by the workflow system.
+ """
+ try:
+ data = event.output if isinstance(event, OutputEvent) else event
Review Comment:
Why do we have this difference?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/message/EventMessage.java:
##########
@@ -20,26 +20,37 @@
import java.util.Objects;
import java.util.OptionalLong;
-/** Specific message type for data. */
-public class EventMessage<T> implements Message {
+/**
+ * Specific message type for event data.
+ *
+ * @param <K> The type of the key.
+ * @param <T> The type of the event.
+ */
+public class EventMessage<K, T> implements Message {
Review Comment:
```suggestion
public class EventMessage<K, E> implements Message {
```
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/PythonRunnerContext.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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.agents.runtime.context;
+
+import org.apache.flink.agents.runtime.message.EventMessage;
+import org.apache.flink.agents.runtime.message.PythonEventMessage;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This class is used to manage the execution context when interacting with Python functions,
+ * including temporarily saving the key of the current Event and collecting new events generated
+ * during the execution of actions.
+ */
+@NotThreadSafe
+public class PythonRunnerContext {
+ private Object key;
+ private List<EventMessage<?, ?>> events;
+
+ public PythonRunnerContext() {
+ this(null);
+ }
+
+ public PythonRunnerContext(Object key) {
+ this.key = key;
+ this.events = new ArrayList<>();
+ }
+
+ public void setKey(Object key) {
+ this.key = key;
+ }
+
+ public void sendEvent(String type, byte[] event) {
+ if (key == null) {
+ throw new IllegalStateException("Key is not set.");
+ }
+ this.events.add(new PythonEventMessage<>(key, event, type));
Review Comment:
Do we need the key for each message?
--
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
Re: [PR] [Feature][runtime] Introducing the FlinkRunnerContext component to support Python workflow execution [flink-agents]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #43:
URL: https://github.com/apache/flink-agents/pull/43#discussion_r2174076190
##########
runtime/src/main/java/org/apache/flink/agents/runtime/PythonActionExecutor.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.agents.runtime;
+
+import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.runtime.context.PythonRunnerContext;
+import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
+import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
+import pemja.core.PythonInterpreter;
+
+import java.util.List;
+
+/** Execute the corresponding Python action in the workflow. */
+public class PythonActionExecutor {
+
+ private static final String IMPORT_FUNCTION = "from flink_agents.plan import function";
+ private static final String IMPORT_FLINK_RUNNER_CONTEXT =
+ "from flink_agents.runtime import flink_runner_context";
+ private static final String IMPORT_PYTHON_JAVA_UTILS =
+ "from flink_agents.runtime import python_java_utils";
+ private static final String CREATE_FLINK_RUNNER_CONTEXT =
+ "flink_runner_context.create_flink_runner_context";
+ private static final String CONVERT_TO_PYTHON_OBJECT =
+ "python_java_utils.convert_to_python_object";
+ private static final String FLINK_RUNNER_CONTEXT_VAR_NAME = "flink_runner_context";
+
+ private final PythonEnvironmentManager environmentManager;
+ private final PythonRunnerContext runnerContext;
+
+ private PythonInterpreter interpreter;
+
+ public PythonActionExecutor(PythonEnvironmentManager environmentManager) {
+ this.environmentManager = environmentManager;
+ this.runnerContext = new PythonRunnerContext();
+ }
+
+ public void open() throws Exception {
+ environmentManager.open();
+ EmbeddedPythonEnvironment env =
+ (EmbeddedPythonEnvironment) environmentManager.createEnvironment();
+
+ interpreter = env.getInterpreter();
+
+ interpreter.exec(IMPORT_FUNCTION);
+ interpreter.exec(IMPORT_FLINK_RUNNER_CONTEXT);
+ interpreter.exec(IMPORT_PYTHON_JAVA_UTILS);
+
+ Object pythonRunnerContextObject =
+ interpreter.invoke(CREATE_FLINK_RUNNER_CONTEXT, runnerContext);
+ interpreter.set(FLINK_RUNNER_CONTEXT_VAR_NAME, pythonRunnerContextObject);
+ }
+
+ public List<PythonEvent> executePythonFunction(PythonFunction function, PythonEvent event)
+ throws Exception {
+ runnerContext.checkNoPendingEvents();
+ function.setInterpreter(interpreter);
+
+ Object pythonRunnerContextObject = interpreter.get(FLINK_RUNNER_CONTEXT_VAR_NAME);
Review Comment:
Why do we need to set / get the context to / from the interpreter?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/PythonActionExecutor.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.agents.runtime;
+
+import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.runtime.context.PythonRunnerContext;
+import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
+import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
+import pemja.core.PythonInterpreter;
+
+import java.util.List;
+
+/** Execute the corresponding Python action in the workflow. */
+public class PythonActionExecutor {
+
+ private static final String IMPORT_FUNCTION = "from flink_agents.plan import function";
+ private static final String IMPORT_FLINK_RUNNER_CONTEXT =
+ "from flink_agents.runtime import flink_runner_context";
+ private static final String IMPORT_PYTHON_JAVA_UTILS =
+ "from flink_agents.runtime import python_java_utils";
+ private static final String CREATE_FLINK_RUNNER_CONTEXT =
+ "flink_runner_context.create_flink_runner_context";
+ private static final String CONVERT_TO_PYTHON_OBJECT =
+ "python_java_utils.convert_to_python_object";
+ private static final String FLINK_RUNNER_CONTEXT_VAR_NAME = "flink_runner_context";
+
+ private final PythonEnvironmentManager environmentManager;
+ private final PythonRunnerContext runnerContext;
+
+ private PythonInterpreter interpreter;
+
+ public PythonActionExecutor(PythonEnvironmentManager environmentManager) {
+ this.environmentManager = environmentManager;
+ this.runnerContext = new PythonRunnerContext();
+ }
+
+ public void open() throws Exception {
+ environmentManager.open();
+ EmbeddedPythonEnvironment env =
+ (EmbeddedPythonEnvironment) environmentManager.createEnvironment();
Review Comment:
Shall we change the signature of `PythonEnvironmentManager#createEnvironment` to avoid this casting?
##########
python/flink_agents/plan/function.py:
##########
@@ -141,3 +141,9 @@ def __call__(self, *args: Tuple[Any, ...], **kwargs: Dict[str, Any]) -> Any:
def check_signature(self, *args: Tuple[Any, ...]) -> None:
"""Check function signature is legal or not."""
+
+
+def call_python_function(module: str, qualname: str, func_args: Tuple[Any, ...]) -> Any:
+ """Used to call a Python function in the Pemja environment."""
+ func = PythonFunction(module=module, qualname=qualname)
Review Comment:
We probably should avoid creating the function over and over again for each call. But let's keep it as is for now, and leave this as a follow-up issue.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/PythonActionExecutor.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.agents.runtime;
+
+import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.runtime.context.PythonRunnerContext;
+import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
+import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
+import pemja.core.PythonInterpreter;
+
+import java.util.List;
+
+/** Execute the corresponding Python action in the workflow. */
+public class PythonActionExecutor {
+
+ private static final String IMPORT_FUNCTION = "from flink_agents.plan import function";
+ private static final String IMPORT_FLINK_RUNNER_CONTEXT =
+ "from flink_agents.runtime import flink_runner_context";
+ private static final String IMPORT_PYTHON_JAVA_UTILS =
+ "from flink_agents.runtime import python_java_utils";
+ private static final String CREATE_FLINK_RUNNER_CONTEXT =
+ "flink_runner_context.create_flink_runner_context";
+ private static final String CONVERT_TO_PYTHON_OBJECT =
+ "python_java_utils.convert_to_python_object";
+ private static final String FLINK_RUNNER_CONTEXT_VAR_NAME = "flink_runner_context";
+
+ private final PythonEnvironmentManager environmentManager;
+ private final PythonRunnerContext runnerContext;
+
+ private PythonInterpreter interpreter;
+
+ public PythonActionExecutor(PythonEnvironmentManager environmentManager) {
+ this.environmentManager = environmentManager;
+ this.runnerContext = new PythonRunnerContext();
+ }
+
+ public void open() throws Exception {
+ environmentManager.open();
+ EmbeddedPythonEnvironment env =
+ (EmbeddedPythonEnvironment) environmentManager.createEnvironment();
+
+ interpreter = env.getInterpreter();
+
+ interpreter.exec(IMPORT_FUNCTION);
+ interpreter.exec(IMPORT_FLINK_RUNNER_CONTEXT);
+ interpreter.exec(IMPORT_PYTHON_JAVA_UTILS);
Review Comment:
Can these importing statements be combined and executed at 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
Re: [PR] [Feature][runtime] Introducing the FlinkRunnerContext component to support Python workflow execution [flink-agents]
Posted by "HuangXingBo (via GitHub)" <gi...@apache.org>.
HuangXingBo merged PR #43:
URL: https://github.com/apache/flink-agents/pull/43
--
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