Can kyuubi work with Airflow's JDBC Operator?

[zhichaoleo]

Hi Team,

I am using Kyuubi to interact with delta lake on Azure, and I plan to call the Hive JDBC URL(jdbc:hive2://<My_Host>:10009) provided by kyuubi through Airflow's JDBC Operator to complete a series of data operations.
However, it doesn't work. My Kyuubi version: 1.3.1-incubating, Airflow version: 2.2.2.

I configure my JDBC connection with below values on airflow:

And my test DAG code is :

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.bash import BashOperator
from airflow.providers.jdbc.operators.jdbc import JdbcOperator
from airflow.providers.jdbc.hooks.jdbc import JdbcHook
from airflow.operators.python import PythonOperator

# Please refer to https://github.com/apache/airflow/blob/main/airflow/providers/jdbc/example_dags/example_jdbc_queries.py

def func(jdbc_conn_id, sql, **kwargs):
    """Print df from JDBC """
    print(kwargs)
    hook = JdbcHook(jdbc_conn_id=jdbc_conn_id)
    df = hook.get_pandas_df(sql=sql)
    print(df.to_string())

with DAG(
    dag_id='kyuubi_jdbc_operator',
    schedule_interval='0 0 * * *',
    start_date=datetime(2022, 1, 1),
    dagrun_timeout=timedelta(minutes=60),
    tags=['example'],
    catchup=False,
) as dag:

    run_this_last = DummyOperator(task_id='run_this_last')

    run_this = PythonOperator(
    task_id='get_data_from_kyuubi',
    python_callable=func,
    op_kwargs={'jdbc_conn_id': 'kyuubi_jdbc', 'sql': '''SELECT * FROM delta.`wasbs://<my_container>@<my_azure_id>.blob.core.windows.net/<table_name>` WHERE PARTITION_DT >= '2022-01-01' AND PARTITION_DT <= '2022-01-13' ''' }, 
    dag=dag,
    )

    run_this >> run_this_last

And there is no error log, just some info logs. Although I turned on the debug log, I still can't see the relevant error message.

*** Reading local file: /home/centos/airflow/logs/kyuubi_jdbc_operator/get_data_from_kyuubi/2022-01-12T06:39:00.206494+00:00/1.log
[2022-01-12, 09:39:02 CST] {base_task_runner.py:63} DEBUG - Planning to run as the  user
[2022-01-12, 09:39:02 CST] {taskinstance.py:720} DEBUG - Refreshing TaskInstance <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> from DB
[2022-01-12, 09:39:02 CST] {taskinstance.py:761} DEBUG - Refreshed TaskInstance <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]>
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Task Instance Not Running' PASSED: True, Task is not in running state.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Not In Retry Period' PASSED: True, The task instance was not marked for retrying.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Task Instance State' PASSED: True, Task state queued was valid.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Trigger Rule' PASSED: True, The task instance did not have any upstream tasks.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1035} INFO - Dependencies all met for <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]>
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Pool Slots Available' PASSED: True, ('There are enough open slots in %s to execute the task', 'default_pool')
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Not In Retry Period' PASSED: True, The task instance was not marked for retrying.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Task Concurrency' PASSED: True, Task concurrency is not set.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]> dependency 'Trigger Rule' PASSED: True, The task instance did not have any upstream tasks.
[2022-01-12, 09:39:02 CST] {taskinstance.py:1035} INFO - Dependencies all met for <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [queued]>
[2022-01-12, 09:39:02 CST] {taskinstance.py:1241} INFO - 
--------------------------------------------------------------------------------
[2022-01-12, 09:39:02 CST] {taskinstance.py:1242} INFO - Starting attempt 1 of 1
[2022-01-12, 09:39:02 CST] {taskinstance.py:1243} INFO - 
--------------------------------------------------------------------------------
[2022-01-12, 09:39:02 CST] {taskinstance.py:1262} INFO - Executing <Task(PythonOperator): get_data_from_kyuubi> on 2022-01-12 06:39:00.206494+00:00
[2022-01-12, 09:39:02 CST] {standard_task_runner.py:52} INFO - Started process 6935 to run task
[2022-01-12, 09:39:02 CST] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'kyuubi_jdbc_operator', 'get_data_from_kyuubi', 'manual__2022-01-12T06:39:00.206494+00:00', '--job-id', '130', '--raw', '--subdir', 'DAGS_FOLDER/kyuubi_jdbc_test.py', '--cfg-path', '/tmp/tmpoq51ew2u', '--error-file', '/tmp/tmpi0gp0tr2']
[2022-01-12, 09:39:02 CST] {standard_task_runner.py:77} INFO - Job 130: Subtask get_data_from_kyuubi
[2022-01-12, 09:39:02 CST] {cli_action_loggers.py:66} DEBUG - Calling callbacks: [<function default_action_log at 0x7f975670b2f0>]
[2022-01-12, 09:39:02 CST] {settings.py:210} DEBUG - Setting up DB connection pool (PID 6935)
[2022-01-12, 09:39:02 CST] {settings.py:267} DEBUG - settings.prepare_engine_args(): Using NullPool
[2022-01-12, 09:39:02 CST] {logging_mixin.py:109} INFO - Running <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]> on host centos
[2022-01-12, 09:39:02 CST] {taskinstance.py:720} DEBUG - Refreshing TaskInstance <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]> from DB
[2022-01-12, 09:39:02 CST] {taskinstance.py:761} DEBUG - Refreshed TaskInstance <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]>
[2022-01-12, 09:39:02 CST] {taskinstance.py:790} DEBUG - Clearing XCom data
[2022-01-12, 09:39:02 CST] {taskinstance.py:797} DEBUG - XCom data cleared
[2022-01-12, 09:39:02 CST] {retries.py:82} DEBUG - Running RenderedTaskInstanceFields._remove_old_rendered_ti_fields_mysql with retries. Try 1 of 3
[2022-01-12, 09:39:02 CST] {taskinstance.py:1429} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=kyuubi_jdbc_operator
AIRFLOW_CTX_TASK_ID=get_data_from_kyuubi
AIRFLOW_CTX_EXECUTION_DATE=2022-01-12T06:39:00.206494+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2022-01-12T06:39:00.206494+00:00
[2022-01-12, 09:39:02 CST] {__init__.py:146} DEBUG - Preparing lineage inlets and outlets
[2022-01-12, 09:39:02 CST] {__init__.py:190} DEBUG - inlets: [], outlets: []
[2022-01-12, 09:39:02 CST] {logging_mixin.py:109} INFO - {'conf': <airflow.configuration.AirflowConfigParser object at 0x7f97579ed9e8>, 'dag': <DAG: kyuubi_jdbc_operator>, 'dag_run': <DagRun kyuubi_jdbc_operator @ 2022-01-12 06:39:00.206494+00:00: manual__2022-01-12T06:39:00.206494+00:00, externally triggered: True>, 'data_interval_end': DateTime(2022, 1, 11, 16, 0, 0, tzinfo=Timezone('UTC')), 'data_interval_start': DateTime(2022, 1, 10, 16, 0, 0, tzinfo=Timezone('UTC')), 'ds': '2022-01-12', 'ds_nodash': '20220112', 'execution_date': <Proxy at 0x7f97537fc948 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f9753706488>>, 'inlets': [], 'logical_date': DateTime(2022, 1, 12, 6, 39, 0, 206494, tzinfo=Timezone('UTC')), 'macros': <module 'airflow.macros' from '/usr/local/lib/python3.6/site-packages/airflow/macros/__init__.py'>, 'next_ds': <Proxy at 0x7f97536ce948 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f9753706510>>, 'next_ds_nodash': <Proxy at 0x7f97536cea48 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f9753706598>>, 'next_execution_date': <Proxy at 0x7f97536ceb48 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f9753706620>>, 'outlets': [], 'params': {}, 'prev_data_interval_start_success': <Proxy at 0x7f97536ce708 with factory <function TaskInstance.get_template_context.<locals>.get_prev_data_interval_start_success at 0x7f975373cea0>>, 'prev_data_interval_end_success': <Proxy at 0x7f97536cec88 with factory <function TaskInstance.get_template_context.<locals>.get_prev_data_interval_end_success at 0x7f975373c950>>, 'prev_ds': <Proxy at 0x7f97536cef88 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f97537066a8>>, 'prev_ds_nodash': <Proxy at 0x7f97536cee88 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f9753706730>>, 'prev_execution_date': <Proxy at 0x7f97536ce748 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f97537067b8>>, 'prev_execution_date_success': <Proxy at 0x7f97536ce988 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f97537068c8>>, 'prev_start_date_success': <Proxy at 0x7f97536cef08 with factory <function TaskInstance.get_template_context.<locals>.get_prev_start_date_success at 0x7f97536f6400>>, 'run_id': 'manual__2022-01-12T06:39:00.206494+00:00', 'task': <Task(PythonOperator): get_data_from_kyuubi>, 'task_instance': <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]>, 'task_instance_key_str': 'kyuubi_jdbc_operator__get_data_from_kyuubi__20220112', 'test_mode': False, 'ti': <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]>, 'tomorrow_ds': <Proxy at 0x7f97536ceec8 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f9753706950>>, 'tomorrow_ds_nodash': <Proxy at 0x7f97536ce5c8 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f97537069d8>>, 'ts': '2022-01-12T06:39:00.206494+00:00', 'ts_nodash': '20220112T063900', 'ts_nodash_with_tz': '20220112T063900.206494+0000', 'var': {'json': None, 'value': None}, 'conn': <airflow.models.taskinstance.TaskInstance.get_template_context.<locals>.ConnectionAccessor object at 0x7f9753703ba8>, 'yesterday_ds': <Proxy at 0x7f97536ce588 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f9753706a60>>, 'yesterday_ds_nodash': <Proxy at 0x7f97536ceb08 with factory <function TaskInstance.get_template_context.<locals>.deprecated_proxy.<locals>.deprecated_func at 0x7f9753706ae8>>, 'templates_dict': None}
[2022-01-12, 09:39:03 CST] {base.py:79} INFO - Using connection to: id: kyuubi_jdbc. Host: jdbc:hive2://<my_host>:10009;transportMode=http;httpPath=cliservice, Port: None, Schema: , Login: ***, Password: ***, extra: {'extra__jdbc__drv_clsname': 'org.apache.hive.jdbc.HiveDriver', 'extra__jdbc__drv_path': '/home/centos/airflow/jars/hive-jdbc-3.1.2.jar'}
[2022-01-12, 09:39:04 CST] {cli_action_loggers.py:84} DEBUG - Calling callbacks: []
[2022-01-12, 09:39:04 CST] {local_task_job.py:154} INFO - Task exited with return code 1
[2022-01-12, 09:39:04 CST] {taskinstance.py:720} DEBUG - Refreshing TaskInstance <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]> from DB
[2022-01-12, 09:39:04 CST] {taskinstance.py:761} DEBUG - Refreshed TaskInstance <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]>
[2022-01-12, 09:39:04 CST] {taskinstance.py:720} DEBUG - Refreshing TaskInstance <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]> from DB
[2022-01-12, 09:39:04 CST] {taskinstance.py:761} DEBUG - Refreshed TaskInstance <TaskInstance: kyuubi_jdbc_operator.get_data_from_kyuubi manual__2022-01-12T06:39:00.206494+00:00 [running]>
[2022-01-12, 09:39:04 CST] {taskinstance.py:2224} DEBUG - Task Duration set to 1.860966
[2022-01-12, 09:39:04 CST] {taskinstance.py:1286} DEBUG - Clearing next_method and next_kwargs.
[2022-01-12, 09:39:04 CST] {taskinstance.py:1280} INFO - Marking task as FAILED. dag_id=kyuubi_jdbc_operator, task_id=get_data_from_kyuubi, execution_date=20220112T063900, start_date=20220112T063902, end_date=20220112T063904
[2022-01-12, 09:39:04 CST] {dagrun.py:619} DEBUG - number of tis tasks for <DagRun kyuubi_jdbc_operator @ 2022-01-12 06:39:00.206494+00:00: manual__2022-01-12T06:39:00.206494+00:00, externally triggered: True>: 2 task(s)
[2022-01-12, 09:39:04 CST] {dagrun.py:634} DEBUG - number of scheduleable tasks for <DagRun kyuubi_jdbc_operator @ 2022-01-12 06:39:00.206494+00:00: manual__2022-01-12T06:39:00.206494+00:00, externally triggered: True>: 1 task(s)
[2022-01-12, 09:39:04 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.run_this_last manual__2022-01-12T06:39:00.206494+00:00 [None]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set.
[2022-01-12, 09:39:04 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.run_this_last manual__2022-01-12T06:39:00.206494+00:00 [None]> dependency 'Not In Retry Period' PASSED: True, The task instance was not marked for retrying.
[2022-01-12, 09:39:04 CST] {taskinstance.py:815} DEBUG - Setting task state for <TaskInstance: kyuubi_jdbc_operator.run_this_last manual__2022-01-12T06:39:00.206494+00:00 [None]> to upstream_failed
[2022-01-12, 09:39:04 CST] {taskinstance.py:1050} DEBUG - <TaskInstance: kyuubi_jdbc_operator.run_this_last manual__2022-01-12T06:39:00.206494+00:00 [upstream_failed]> dependency 'Trigger Rule' PASSED: False, Task's trigger rule 'all_success' requires all upstream tasks to have succeeded, but found 1 non-success(es). upstream_tasks_state={'total': 1, 'successes': 0, 'skipped': 0, 'failed': 1, 'upstream_failed': 0, 'done': 1}, upstream_task_ids={'get_data_from_kyuubi'}
[2022-01-12, 09:39:04 CST] {taskinstance.py:1029} DEBUG - Dependencies not met for <TaskInstance: kyuubi_jdbc_operator.run_this_last manual__2022-01-12T06:39:00.206494+00:00 [upstream_failed]>, dependency 'Trigger Rule' FAILED: Task's trigger rule 'all_success' requires all upstream tasks to have succeeded, but found 1 non-success(es). upstream_tasks_state={'total': 1, 'successes': 0, 'skipped': 0, 'failed': 1, 'upstream_failed': 0, 'done': 1}, upstream_task_ids={'get_data_from_kyuubi'}
[2022-01-12, 09:39:04 CST] {local_task_job.py:264} INFO - 0 downstream tasks scheduled from follow-on schedule check

Could you help me with this problem?

Regards & Thanks

[yaooqinn]

you might need to downgrade hive jdbc from 3.1.2 to 2.3.9 or lower.

Also, it's helpful to check the kyuubi server and engine's log

[zhichaoleo]

@yaooqinn Thanks for your quick reply. Will have a try.