当使用Airflow与Databricks笔记本一起使用时，on_success_callback和on_failure_callback可能无法正常工作的原因是Databricks笔记本任务的状态不会触发Airflow的回调函数。为了解决这个问题，可以使用以下方法：

1. 使用Databricks提供的DatabricksRunNowOperator来替代DatabricksSubmitRunOperator。

from airflow.contrib.operators.databricks_operator import DatabricksRunNowOperator

run_notebook_task = DatabricksRunNowOperator(
    task_id='run_notebook_task',
    job_id='job_id',
    notebook_params={'param1': 'value1'},
    dag=dag
)

2. 在Databricks笔记本中自定义处理成功和失败的逻辑，并将结果保存到Databricks表中。

# 在Databricks笔记本中定义成功和失败的处理逻辑
if success_condition:
    status = 'success'
else:
    status = 'failure'

# 将结果保存到Databricks表中
dbutils.notebook.exit(status)

3. 在Airflow中添加一个新的任务，通过DatabricksSubmitRunOperator检查Databricks笔记本的状态并触发回调函数。

from airflow.contrib.operators.databricks_operator import DatabricksSubmitRunOperator
from airflow.operators.python_operator import PythonOperator

def check_notebook_status():
    # 检查Databricks笔记本的状态
    # 如果成功，则触发on_success_callback
    # 如果失败，则触发on_failure_callback

check_status_task = PythonOperator(
    task_id='check_status_task',
    python_callable=check_notebook_status,
    dag=dag
)

run_notebook_task >> check_status_task

通过这种方法，您可以在Databricks笔记本任务完成后，通过自定义的逻辑检查其状态，并触发相应的回调函数。