yuxiqian commented on code in PR #4501: URL: https://github.com/apache/flink-cdc/pull/4501#discussion_r3802266745
########## flink-cdc-python/src/main/java/org/apache/flink/cdc/python/PythonUdf.java: ########## @@ -0,0 +1,226 @@ +/* + * 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.cdc.python; + +import org.apache.flink.cdc.common.annotation.Experimental; +import org.apache.flink.cdc.common.configuration.ConfigOption; +import org.apache.flink.cdc.common.configuration.ConfigOptions; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.common.udf.UserDefinedFunction; +import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext; +import org.apache.flink.cdc.python.utils.PythonUdfSignature; + +import pemja.core.PythonInterpreter; +import pemja.core.PythonInterpreterConfig; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Locale; +import java.util.stream.Stream; +import java.util.zip.ZipEntry; +import java.util.zip.ZipInputStream; + +/** Generic UDF that delegates to a Python function defined inline in YAML. */ +@Experimental +public final class PythonUdf implements UserDefinedFunction { + + public static final ConfigOption<String> OPTION_SOURCE = + ConfigOptions.key("source") + .stringType() + .noDefaultValue() + .withDescription("Inline Python source containing a `def eval(...)`."); + + public static final ConfigOption<String> OPTION_PYTHON_EXECUTABLE = + ConfigOptions.key("python-executable") + .stringType() + .defaultValue("python3") + .withDescription( + "Path to the Python interpreter Pemja embeds on every TaskManager." + + " The interpreter must have a matching `pemja` package" + + " installed; defaults to the first `python3` on PATH."); + + public static final ConfigOption<String> OPTION_PYTHON_FILES = + ConfigOptions.key("python-files") + .stringType() + .noDefaultValue() + .withDescription( + "Comma-separated directories or zip archives that will be added" + + " to the embedded Python import search path. Zip archives" + + " are extracted to a temporary directory first so packages" + + " with native extensions can be imported."); + + private static final String PYTHON_FUNCTION_NAME = "eval"; + + private transient PythonInterpreter interpreter; + private transient Path extractedPythonFilesDirectory; + + @Override + public void open(UserDefinedFunctionContext context) { + Configuration config = context.configuration(); + String source = requireSource(config); + String pythonExec = config.get(OPTION_PYTHON_EXECUTABLE); + + PythonInterpreterConfig.PythonInterpreterConfigBuilder pemjaConfigBuilder = + PythonInterpreterConfig.newBuilder().setPythonExec(pythonExec); + configurePythonFiles(pemjaConfigBuilder, config); + + this.interpreter = new PythonInterpreter(pemjaConfigBuilder.build()); + this.interpreter.exec(source); Review Comment: Fixed. But we may have no chance to clean-up resources if native error occurs and JVM crashes. As `java.io.tmpdir` usually points to `/tmp` where files are not written to disks, it might be acceptable. -- 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]
