Airflow GrpcOperator中Protobuf请求消息模板化失效问题求助

问题根源：直接在GrpcOperator的data字段中实例化Protobuf消息对象时，该对象会在DAG解析阶段被创建，此时Airflow的模板引擎还未渲染{{ ds }}这类变量，导致变量以原始字符串形式传入请求。

以下是两种可行的解决办法：

方法1：自定义GrpcOperator子类，动态生成请求对象 #

重写prepare_request方法，在任务运行阶段（模板变量已渲染完成后）再构造Protobuf请求：

from airflow.providers.grpc.operators.grpc import GrpcOperator
import proto_pb2  # 替换为你的Protobuf模块路径

class TemplatedGrpcOperator(GrpcOperator):
    def prepare_request(self, context):
        # 渲染data中的模板变量
        rendered_data = self.render_template(self.data, context)
        # 动态生成Protobuf请求对象
        request = proto_pb2.CustomGrpcServiceRequest(
            request_date=rendered_data["request_date"]
        )
        return {"request": request}

# 使用自定义Operator
task_grpc = TemplatedGrpcOperator(
    dag=dag,
    task_id="task_grpc",
    grpc_conn_id="grpc_default",
    stub_class=CustomGrpcServiceStub,
    call_func="CustomGrpcServiceFunction",
    response_callback=CustomGrpcService_callback,
    streaming=False,
    data={"request_date": "{{ ds }}"}  # 仅传递需模板化的字段
)

方法2：使用PythonOperator完全控制请求流程 #

放弃GrpcOperator，改用PythonOperator直接编写gRPC调用逻辑，更灵活可控：

from airflow.operators.python import PythonOperator
import grpc
import proto_pb2
import proto_pb2_grpc  # 替换为你的Protobuf服务模块路径

def call_grpc_service(**context):
    # 获取渲染后的DAG运行日期
    ds = context["ds"]
    # 建立gRPC连接（也可通过Airflow连接管理获取地址，这里简化示例）
    with grpc.insecure_channel("your_grpc_server_host:port") as channel:
        stub = proto_pb2_grpc.CustomGrpcServiceStub(channel)
        # 构造带模板变量的请求
        request = proto_pb2.CustomGrpcServiceRequest(request_date=ds)
        # 调用gRPC服务
        response = stub.CustomGrpcServiceFunction(request)
        # 执行回调逻辑
        CustomGrpcService_callback(response)

task_grpc = PythonOperator(
    dag=dag,
    task_id="task_grpc",
    python_callable=call_grpc_service,
    provide_context=True
)

这两种方法都能确保模板变量在任务运行阶段被正确渲染后，再传入Protobuf请求对象中。

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