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