要实现“Airflow仅在工作日进行补偿”，可以使用Airflow的BranchPythonOperator和PythonOperator来编写自定义的任务控制逻辑。

以下是一个示例代码，演示如何在工作日进行补偿任务：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python_operator import BranchPythonOperator, PythonOperator

def check_workday(**kwargs):
    current_date = datetime.now().date()
    # 假设周六和周日为非工作日
    if current_date.weekday() < 5:
        return "do_compensation_task"
    else:
        return "no_compensation_task"

def do_compensation_task(**kwargs):
    # 执行补偿任务的逻辑
    print("Performing compensation task on workday")

def no_compensation_task(**kwargs):
    # 非工作日不执行补偿任务的逻辑
    print("No compensation task on non-workday")

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2022, 1, 1),
    'retries': 0
}

dag = DAG('compensation_dag', schedule_interval='@daily', default_args=default_args)

check_workday_task = BranchPythonOperator(
    task_id='check_workday',
    python_callable=check_workday,
    provide_context=True,
    dag=dag
)

do_compensation_task = PythonOperator(
    task_id='do_compensation_task',
    python_callable=do_compensation_task,
    provide_context=True,
    dag=dag
)

no_compensation_task = PythonOperator(
    task_id='no_compensation_task',
    python_callable=no_compensation_task,
    provide_context=True,
    dag=dag
)

check_workday_task >> [do_compensation_task, no_compensation_task]

在上述代码中，check_workday函数用于检查当前日期是否为工作日。如果是工作日，它返回任务ID 'do_compensation_task'，否则返回任务ID 'no_compensation_task'。

然后，使用BranchPythonOperator将该函数作为任务添加到DAG中。根据返回的任务ID，DAG将决定执行补偿任务还是非补偿任务。

do_compensation_task和no_compensation_task分别是PythonOperator，用于执行补偿任务和非补偿任务的逻辑。

最后，将check_workday_task与do_compensation_task和no_compensation_task使用>>操作符连接起来，以定义任务之间的依赖关系。