Airflow导入DAG方法及自定义Jar DAG导入异常排查

Hey there! Let's tackle your Airflow issues step by step, starting with getting your DAG to show up in the UI, then covering how to properly import DAGs in Airflow.

First: Fixing Your DAG Not Showing Up in the UI #

Looking at your code and the problem description, here are the key issues and fixes to check:

1. Your Java Command Is Incomplete #

The t2 task's bash command only loads the JAR but doesn't execute any code:

java -cp /home/user/Docs/JObforAirflow/test/jobbatch.jar

Talend-generated JARs require you to specify the main class (or use -jar if it's an executable JAR). For example:

• If it's an executable JAR: java -jar /home/user/Docs/JObforAirflow/test/jobbatch.jar
• If you need to specify the main class: java -cp /home/user/Docs/JObforAirflow/test/jobbatch.jar org.talend.daikon.runtime.Main (replace with your actual Talend main class)

A broken command like your original will cause the DAG to fail parsing, so Airflow will skip loading it entirely.

2. Task Dependency Logic Might Be Reversed #

You wrote t1.set_upstream(t2), which means t2 runs before t1. If you intended to run the dependency script (t1) first, then the JAR (t2), use this instead:

t1 >> t2  # Airflow 2.x+ recommended syntax, more readable
# Or the older syntax: t2.set_upstream(t1)

3. Start Date Is Too Old #

Your start_date is set to 2019, which is far in the past. If Airflow's catchup setting is disabled (default in newer versions), it might hide DAGs with past start dates. Update the start date to a recent date (e.g., datetime(2024, 1, 1)) and add catchup=False to your default_args to avoid backfilling old runs.

4. General Troubleshooting Steps #

If the above fixes don't work, try these:

• Check DAG file location: Run airflow config get-value core dags_folder to confirm your .py file is in the correct directory.
• Test DAG parsing manually: Run python your_dag_file.py in the terminal—any syntax errors or runtime exceptions here will prevent Airflow from loading the DAG.
• Check scheduler logs: Look in $AIRFLOW_HOME/logs/scheduler for error messages about your DAG.
• Restart Airflow services: If you just added the DAG file, restart the webserver and scheduler:

  airflow webserver restart
  airflow scheduler restart
  

• Refresh the UI: Click the "Refresh" button in the Airflow UI to reload the DAG list.

Second: How to Import DAGs in Airflow #

Airflow automatically loads DAGs—you don't need to "import" them manually via terminal commands. Follow these steps:

1. Write valid DAG code: Ensure your .py file defines a DAG object, has no syntax errors, and runs without exceptions (test with python your_dag.py).
2. Place the file in the DAGs folder: Move your .py file to the directory specified by airflow config get-value core dags_folder.
3. Verify permissions: Make sure the Airflow user (running the webserver/scheduler) has read access to the file.
4. Wait or refresh: Airflow's scheduler scans the DAGs folder every 30 seconds by default. If you don't want to wait, restart the services or click "Refresh" in the UI.
5. Confirm the DAG exists: Run airflow dags list in the terminal to check if your DAG is listed—if it is but doesn't show in the UI, clear your browser cache or hard-refresh the page.

Fixed Example of Your DAG Code #

Here's a cleaned-up version of your code with the fixes applied:

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime

# Ensure paths are correct and executable
bib_app = "/home/user/Docs/JObforAirflow/test/test_run.sh"
talend_jar_path = "/home/user/Docs/JObforAirflow/test/jobbatch.jar"
# Replace with your actual Talend main class
talend_main_class = "org.talend.daikon.runtime.Main"

default_args = {
    'owner': 'yabid',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),  # Recent start date
    'email': ['user@user.com'],
    'email_on_failure': True,
    'email_on_success': True,
    'provide_context': True,
    'catchup': False  # Disable backfilling past runs
}

# Define DAG with no scheduled interval (run manually)
dag = DAG('run_jar', default_args=default_args, schedule_interval=None)

t1 = BashOperator(task_id='dependency_setup', bash_command=bib_app, dag=dag)
t2 = BashOperator(
    task_id='execute_talend_jar',
    dag=dag,
    # Use executable JAR syntax or specify main class
    bash_command=f'java -jar {talend_jar_path}'
    # Or: bash_command=f'java -cp {talend_jar_path} {talend_main_class}'
)

# Set correct execution order: t1 runs first, then t2
t1 >> t2

内容的提问来源于stack exchange，提问作者Yassin Abid