qflink-python/ffi-demo/
├── src/main/java/cn/syntomic/ffi/FFIDemo.java   # Java：声明 native 方法
├── src/main/python/day_of_week.py               # Python：真正干活的函数
├── src/main/c/ffi_demo.c                        # C：JNI + Python/C API 的桥
└── run.sh                                       # 编译 Java 和 C，然后运行

qflink-python/ffi-demo/run.sh
qflink-python/ffi-demo/run.sh 2026-10-24

4
6

package cn.syntomic.ffi;

/** Java 传一个日期给 Python，Python 告诉我们这天星期几。 */
public class FFIDemo {

    static {
        // 加载用 C 写的“桥”。加载时 JVM 会调用其中的 JNI_OnLoad，Python 解释器就在那里启动
        System.load(System.getProperty("user.dir") + "/" + System.mapLibraryName("ffidemo"));
    }

    /** native 方法：Java 这里只有声明，实现写在 C 里 */
    native int dayOfWeek(String date);

    public static void main(String[] args) throws InterruptedException {
        FFIDemo demo = new FFIDemo();
        System.out.println(demo.dayOfWeek(args.length > 0 ? args[0] : "1994-05-05"));
    }
}

javac -h src/main/c -d target/classes src/main/java/cn/syntomic/ffi/FFIDemo.java

JNIEXPORT jint JNICALL Java_cn_syntomic_ffi_FFIDemo_dayOfWeek
  (JNIEnv *, jobject, jstring);

from datetime import datetime


def day_of_week(date: str) -> int:
    """'1994-05-05' -> 4（星期一是 1，星期日是 7）"""
    return datetime.strptime(date, "%Y-%m-%d").isoweekday()

#include <Python.h>
#include "cn_syntomic_ffi_FFIDemo.h"

static PyObject *day_of_week; // 导入一次、反复使用的 Python 函数

/* System.load 加载这个库时，JVM 会先调用 JNI_OnLoad：解释器在这里启动，只启动一次 */
JNIEXPORT jint JNICALL JNI_OnLoad(JavaVM *vm, void *reserved) {
    Py_Initialize();
    PyRun_SimpleString("import sys; sys.path.insert(0, 'src/main/python')");

    PyObject *module = PyImport_ImportModule("day_of_week");
    day_of_week = module ? PyObject_GetAttrString(module, "day_of_week") : NULL;
    Py_XDECREF(module);
    if (day_of_week == NULL) {
        PyErr_Print();
        return JNI_ERR;
    }

    // Py_Initialize 之后，当前线程一直拿着 GIL；交出来，其他线程才能调用 Python
    PyEval_SaveThread();
    return JNI_VERSION_1_8;
}

JNIEXPORT jint JNICALL Java_cn_syntomic_ffi_FFIDemo_dayOfWeek(JNIEnv *env, jobject self,
                                                              jstring date) {
    // ① Java 的 String -> C 的字符串
    const char *text = (*env)->GetStringUTFChars(env, date, NULL);

    // ② 拿到 GIL：同一时刻，只有拿着它的线程能执行 Python
    PyGILState_STATE gil = PyGILState_Ensure();

    // ③ C 字符串 -> Python 的 str，调用 day_of_week(text)
    PyObject *result = PyObject_CallFunction(day_of_week, "s", text);

    // ④ Python 的 int -> C 的 long；Python 抛了异常就转成 Java 异常
    long weekday = -1;
    if (result != NULL) {
        weekday = PyLong_AsLong(result);
        Py_DECREF(result);
    } else {
        PyErr_Print();
        (*env)->ThrowNew(env, (*env)->FindClass(env, "java/lang/RuntimeException"),
                         "Python 函数执行失败");
    }

    // ⑤ 归还 GIL，释放字符串
    PyGILState_Release(gil);
    (*env)->ReleaseStringUTFChars(env, date, text);
    return (jint)weekday;
}

py() { "$PYTHON" -c "import sysconfig; print(sysconfig.$1)"; }
py_include="$(py 'get_paths()["include"]')"       # Python.h 所在的目录
py_libdir="$(py 'get_config_var("LIBDIR")')"       # libpython 所在的目录
py_version="$(py 'get_config_var("LDVERSION")')"   # 比如 3.12

# 1. Java：编译，并为 native 方法生成 C 头文件
javac -h src/main/c -d target/classes src/main/java/cn/syntomic/ffi/FFIDemo.java

# 2. C：头文件来自 JDK 和 Python，链接 libpython，产出动态库
cc -shared -fPIC \
    -I"$JAVA_HOME/include" -I"$JAVA_HOME/include/$platform" \
    -I"$py_include" \
    src/main/c/ffi_demo.c \
    -L"$py_libdir" -lpython"$py_version" -Wl,-rpath,"$py_libdir" \
    -o "$library"

# 3. 运行：Java -> JNI -> C -> Python/C API -> Python
java -cp target/classes cn.syntomic.ffi.FFIDemo "$@"

static void run(FFIDemo demo, int threads, int total) throws InterruptedException {
    List<Thread> workers = new ArrayList<>();
    for (int i = 0; i < threads; i++) {
        Thread worker =
                new Thread(
                        () -> {
                            for (int n = 0; n < total / threads; n++) {
                                demo.dayOfWeek("1994-05-05");
                            }
                        });
        workers.add(worker);
        worker.start();
    }
    for (Thread worker : workers) {
        worker.join();
    }
}

qflink-python/ffi-demo/run.sh bench

400000 次调用
1 个线程：1.49 秒
2 个线程：2.14 秒
4 个线程：2.36 秒

PythonInterpreterConfig config = PythonInterpreterConfig.newBuilder()
        .setPythonExec("python3")
        .addPythonPaths(path)
        .build();
PythonInterpreter interpreter = new PythonInterpreter(config);

interpreter.exec("import str_upper");
Object result = interpreter.invoke("str_upper.upper", "abcd"); // "ABCD"
interpreter.close();

// open()：每个算子实例创建自己的 PythonInterpreter，把 UDF 准备好
interpreter = new PythonInterpreter(config);
interpreter.exec("from pyflink.fn_execution.embedded.operation_utils"
        + " import create_scalar_operation_from_proto");
interpreter.set("proto", udfProto); // 描述 UDF 的 protobuf 字节，只传一次
interpreter.exec("scalar_operation = create_scalar_operation_from_proto(proto, ...)");
interpreter.invokeMethod("scalar_operation", "open");

// processElement()：每来一条数据，在算子线程里同步调用一次
Object args = inputConverter.toExternal(row); // Flink 内部格式 -> Java 对象
Object result = interpreter.invokeMethod("scalar_operation", "process_element", args);
output.collect(outputConverter.toInternal(result)); // Java 对象 -> Flink 内部格式

SET 'python.execution-mode' = 'thread';

t_env.get_config().set("python.execution-mode", "thread")

tutorials/run.sh --python sql_is_all_your_need/flink_sql_udf/python_udf.sql