[Feature][Transform] Add python script tool to Transform

[liugddx]

Search before asking

• I had searched in the feature and found no similar feature requirement.

Description

Clean data by writing python scripts.

transform {
Python {
for key, value in input_dict.items():
if key == "name":
cleaned_value = value.strip() if isinstance(value, str) else value
elif key == "age":
try:
cleaned_value = int(value)
except ValueError:
cleaned_value = None
elif key == "email":
cleaned_value = value.lower() if isinstance(value, str) else value
else:
cleaned_value = value
result.append(cleaned_value)
}
}

Usage Scenario

No response

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Hisoka-X]

+1

[BruceWong96]

+1

[FuYouJ]

Please assign it to me

[FuYouJ]

I looked into a solution using py4j, and here's how it works. It supports bidirectional communication between java and python clients.
To fulfill this requirement:
1：During the startup of the Seatunnel transform, a resident process is initiated from the Java side. The code is as follows

GatewayServer gatewayServer = new GatewayServer(new ExampleJava());
gatewayServer.start();

ExampleJava provides the following encapsulation of basic SeatunnelRow operations。
2：Now that we have started a Java server, we need to launch a Python client to connect to the Java server

 gateway = JavaGateway(
        callback_server_parameters=CallbackServerParameters()
    )

    # 获取 Java 对象 其实就是ExampleJava2
    java_app = gateway.entry_point

3: How to execute python code in java?
Suppose we have a getPython() method on the java side that provides user-configured python code

 public String getPythonCode(){
        return "import numpy as np;\n" +
                "print(np.array([1, 2, 3]));print(\"[Python] 这是 Java 传入的代码！java_app.callPythonMethod(1, 1.1)\");java_app.callPythonMethod(1, 1.1)";
    }

 # 在python 处申明java对象
        java_list = gateway.jvm.java.util.ArrayList()
        java_list.append(1)
        java_list.append(2)
        java_app.addList(1,java_list)
        # 在python 处申明java对象
        my_object = gateway.jvm.com.example.MyObject("test", 123)
        java_app.addObject(1,my_object);
        try:
            ## 执行 java段给定的自定义代码
            exec(java_app.getPythonCode())  # 动态执行 Python 代码
        except Exception as e:
            print(f"[Python] 代码执行出错: {e}")

4: The flow chart is as follows

5: code

package com.example;

import py4j.GatewayServer;

import java.util.Collection;
import java.util.List;

/**
 * {@code Author} FuYouJ
 * {@code Date} 2025/2/1 19:44
 */

public class ExampleJava2 {
    private PythonCallbackInterface2 callback;

    // 注册 Python 回调对象
    public void registerCallback(PythonCallbackInterface2 callback) {
        this.callback = callback;
    }

    // Java 主动调用 Python 的方法
    public void callPythonMethod(Integer index,String val) {
        if (callback != null) {
            System.out.println("接收到"+ index + "的值"+ val);
        } else {
            System.out.println("[Java] Python 回调未注册！");
        }
    }
    public void addList(Integer index, Collection<Integer> collection){
        System.out.println("接收到collection");
        for (Integer s : collection) {
            System.out.println(s);
        }
    }

    public void callPythonMethod(Integer index,Double val) {
        if (callback != null) {
            System.out.println("接收到"+ index + "的值"+ val);
        } else {
            System.out.println("[Java] Python 回调未注册！");
        }
    }

    public void addObject(Integer index,MyObject object){
        System.out.println("接收到MyObject");
        System.out.println(object.toString());
    }

    int messageNum = 0;

    // 被 Python 调用的 Java 方法
    public String getJavaMessage() {
        return "Hello from Java!" + messageNum ++;
    }

    public String getPythonCode(){
        return "import numpy as np;\n" +
                "print(np.array([1, 2, 3]));print(\"[Python] 这是 Java 传入的代码！java_app.callPythonMethod(1, 1.1)\");java_app.callPythonMethod(1, 1.1)";
    }

    public int add(int a, int b) {
        return a + b;
    }

    public String getMessage() {
        return "来自Java语言的问候";
    }

    public static void main(String[] args) {
        GatewayServer gatewayServer = new GatewayServer(new ExampleJava2());
        gatewayServer.start();
        System.out.println("[Java] Gateway Server started on port " + gatewayServer.getPort());
    }

    class MyObject {
        private String name;
        private int value;


        public MyObject(String name, int value) {
            this.name = name;
            this.value = value;
        }

        public String getName() {
            return name;
        }

        public int getValue() {
            return value;
        }

        @Override
        public String toString() {
            return "MyObject{name='" + name + "', value=" + value + "}";
        }
    }
}

python

from py4j.java_gateway import JavaGateway, CallbackServerParameters

# 实现 Java 接口的 Python 类
class PythonCallback:
    def processArray(self, index, val):
        # 处理 index 和 val
        print(f"[Python] 接收到 Java 传递的数据: index={index}, val={val}")

    class Java:
        implements = ["com.example.PythonCallbackInterface2"]    # 声明实现 Java 接口

def main():
    # 连接到 Java 服务端（默认 25333），并启动 Python 回调服务端（默认 25334）
    gateway = JavaGateway(
        callback_server_parameters=CallbackServerParameters()
    )

    # 获取 Java 对象 其实就是ExampleJava2
    java_app = gateway.entry_point

    # 注册 Python 回调到 Java
    python_callback = PythonCallback()
    java_app.registerCallback(python_callback)
    for i in range(10):
        # 测试 Python 调用 Java 方法
        # print("[Python] Java 消息:", java_app.getJavaMessage())
        # # 触发 Java 主动调用 Python 方法
        # print("[Python] 要求 Java 调用 Python...")
        # java_app.callPythonMethod(1, 1.1)  # Java 会在此处调用 Python 的 processArray() 方法
        # print(f"[Python] 执行 Java 传入的代码: {java_app.getPythonCode()}")
        # 在python 处申明java对象
        java_list = gateway.jvm.java.util.ArrayList()
        java_list.append(1)
        java_list.append(2)
        java_app.addList(1,java_list)
        # 在python 处申明java对象
        my_object = gateway.jvm.com.example.MyObject("test", 123)
        java_app.addObject(1,my_object);
        try:
            ## 执行 java段给定的自定义代码
            exec(java_app.getPythonCode())  # 动态执行 Python 代码
        except Exception as e:
            print(f"[Python] 代码执行出错: {e}")

if __name__ == "__main__":
    main()

You must start java before starting python

[FuYouJ]

video show
http://www.xiewenshi.fans/liaomingtao/%E5%BD%95%E5%B1%8F2025-02-01%2021.26.32.mov

[FuYouJ]

@liugddx @hailin0 @Hisoka-X

Give me some advice and I'm ready to do it

[csh0101]

LGTM

[hailin0]

Python transform may also need to output CatalogTable structure changes

[liugddx]

Are there any prerequisites for using py4j?

[SimonChou12138]

LGTM

[FuYouJ]

Are there any prerequisites for using py4j?

java端 seatunnel会自己引入
python端 需要服务器提供python环境