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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
