如何使用JDBC操作符在airflow中获取sql查询结果

from airflow.models import DAG
from airflow.providers.jdbc.operators.jdbc import JdbcOperator
from datetime import datetime

dag = DAG(
    dag_id='jdbc_operator_example',
    start_date=datetime(2022, 1, 1),
    schedule_interval=None
)

jdbc_operator = JdbcOperator(
    task_id='jdbc_task',
    jdbc_conn_id='jdbc_connection',
    sql="SELECT * FROM your_table",
    dag=dag
)

from airflow.hooks.base import BaseHook
from airflow.utils.file import TemporaryDirectory, get_task_logger

class JdbcToFileOperator(JdbcOperator):
    def execute(self, context):
        with TemporaryDirectory(prefix='jdbc_to_file_') as tmp_dir:
            hook = BaseHook.get_hook(conn_id=self.jdbc_conn_id)
            connection = hook.get_conn()

            cursor = connection.cursor()
            cursor.execute(self.sql)
            result = cursor.fetchall()

            logger = get_task_logger(__name__)
            file_path = tmp_dir + "/result.txt"
            with open(file_path, 'w') as file:
                for row in result:
                    file.write(str(row) + '\n')

            logger.info(f"Saved query result to {file_path}")

jdbc_to_file_operator = JdbcToFileOperator(
    task_id='jdbc_to_file_task',
    jdbc_conn_id='jdbc_connection',
    sql="SELECT * FROM your_table",
    dag=dag
)

jdbc_operator >> jdbc_to_file_operator