Airflow 2.1.0如何直接调用指定Java类或方法无需单独打包jar

下面给出两种不需要为每段业务逻辑单独打jar包的实现方式：

方案1：复用统一Jar包，通过BashOperator指定执行类 #

你不需要单独给每个业务逻辑打包可执行jar，只需要把所有业务代码、依赖包统一打包成一个普通jar，或者统一放到同一个依赖目录，执行时通过-cp参数指定classpath，后跟要执行的全限定类名即可，还支持自定义传参。
示例代码如下：

from airflow.operators.bash import BashOperator

call_java_class_task = BashOperator(
    task_id='call_specified_java_class',
    # -cp参数后接统一jar包路径/依赖目录，多个路径用:分隔（Linux环境）或;分隔（Windows环境），之后是要执行的全限定类名，末尾可加自定义业务参数
    bash_command='java -cp /opt/business/unified.jar:/opt/business/dependency/* com.yourcompany.business.DataValidator {{ params.input_file_path }}',
    params={'input_file_path': '/data/source/20240520.csv'},
    dag=dag
)

该方案改造成本极低，你只需要为每个业务逻辑编写一个带main方法的Java类即可，所有代码都可以放到同一个统一jar包中维护。

方案2：通过JPype实现在Python中直接调用任意Java方法 #

如果不想为每个业务逻辑编写main方法，可以用JPype在Python代码中直接调用Java类的任意方法，还能直接获取方法返回值同步到Airflow的XCom中供下游任务使用，灵活度更高。

前置依赖 #

先在所有Airflow Worker节点安装JPype：
pip install JPype1==1.4.1（注意版本要和Worker节点的JDK版本适配）

示例代码如下：

from airflow.operators.python import PythonOperator
import jpype
from jpype.types import *

def call_java_logic(**context):
    # 启动JVM，指定JVM路径和包含业务代码的classpath
    jpype.startJVM(
        jpype.getDefaultJVMPath(),
        "-Djava.class.path=/opt/business/unified.jar:/opt/business/dependency/*",
        convertStrings=False
    )

    # 导入要调用的Java类
    DataCleaner = jpype.JClass("com.yourcompany.business.DataCleaner")
    # 既可以调用静态方法，也可以实例化类之后调用普通方法
    clean_result = DataCleaner.cleanRawData(context["params"]["raw_data"])
    # 方法返回值可直接存入XCom供下游任务使用
    context["ti"].xcom_push(key="clean_result", value=str(clean_result))

    # 任务执行完成关闭JVM
    jpype.shutdownJVM()

direct_call_java_task = PythonOperator(
    task_id="direct_call_java_method",
    python_callable=call_java_logic,
    params={"raw_data": "待处理的原始业务数据"},
    dag=dag
)

注意事项 #

• 两种方案都要求Airflow Worker节点已经安装适配版本的JDK，且环境变量配置正确，可正常执行java命令
• 如果使用JPype，注意同一个Python进程中JVM只能启动一次，多个Java调用任务建议分配独立的Worker进程避免冲突
• 业务代码迭代后只需要替换Worker节点上的统一jar包即可，无需修改DAG代码（除非类名、方法签名发生变更）

内容的提问来源于stack exchange，提问作者Shashank Gupta