使用Apache Airflow和Python编写DAG任务，其中有不同的操作符和触发器可以实现非线性DAG执行。

示例代码：

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2021, 1, 1),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'non_linear_dag_example',
    default_args=default_args,
    description='A non-linear DAG example',
    schedule_interval=timedelta(days=1)
)

start = DummyOperator(task_id='start', dag=dag)

task1 = BashOperator(
    task_id='task1',
    bash_command='echo "Task 1"',
    dag=dag,
)

task2 = BashOperator(
    task_id='task2',
    bash_command='echo "Task 2"',
    dag=dag,
)

task3 = BashOperator(
    task_id='task3',
    bash_command='echo "Task 3"',
    dag=dag,
)

task4 = BashOperator(
    task_id='task4',
    bash_command='echo "Task 4"',
    dag=dag,
)

end = DummyOperator(task_id='end', dag=dag)

start >> task1 >> [task2, task3] >> task4 >> end
task2 << task3

上述代码中使用了DummyOperator和BashOperator来创建任务，它们可以实现DAG的非线性执行。任务图如下所示：

start -> task1 -> task2 -> task4 -> end
         \-> task3 -/

其中，start、task2、task4和end是DummyOperator的实例，而task1和task3则是BashOperator的实例。在任务图中，箭头指向的任务需要先于指向箭头的任务执行，当一个