How Flink and Airflow Work Together

1. Airflow as a Flink Job Orchestrator: The Primary Model

This is the most common and recommended way to integrate the two.

Mechanism Breakdown:

• FlinkOperator: Airflow’s FlinkOperator (provided by the Apache Flink provider) is the primary tool. It handles the complexities of interacting with a Flink cluster.
• Connection Management: You configure a Flink connection in Airflow, specifying the JobManager’s host and port. The FlinkOperator uses this connection to communicate with the cluster.
• Job Submission: You specify the path to your Flink application JAR file (application parameter), the job name (job_name), and any application-specific arguments (application_args).
• Configuration: You can pass Flink-specific configurations using the conf parameter (a dictionary of key-value pairs).
• Parallelism Control: The parallelism parameter allows you to set the initial parallelism of the Flink job.
• Monitoring: The FlinkOperator periodically polls the Flink JobManager’s REST API to track the job’s status. The Airflow task will remain in a ‘running’ state until the Flink job completes (successfully or with failure).
• Dependency Management: Airflow ensures that the Flink job is only submitted after all its upstream tasks in the DAG have successfully completed.
• Failure Handling: If the Flink job fails, the FlinkOperator will mark the Airflow task as failed, allowing Airflow’s retry mechanisms and alerting to take effect.

Example (Detailed):

from airflow import DAG
from airflow.providers.apache.flink.operators.flink import FlinkOperator
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

with DAG(
    dag_id='detailed_flink_pipeline',
    start_date=days_ago(2),
    schedule_interval='@once',
    catchup=False,
    tags=['flink', 'orchestration'],
) as dag:
    prepare_data = BashOperator(
        task_id='prepare_input_data',
        bash_command='echo "Preparing data..." && sleep 10',
    )

    submit_realtime_flink_job = FlinkOperator(
        task_id='submit_realtime_analytics',
        flink_conn_id='flink_cluster',
        application='/opt/flink/usertools/examples/StreamingJoins.jar',
        job_name='realtime-user-activity-join',
        parallelism=8,
        conf={
            'flink.checkpointing.interval': '60000',
            'pipeline.max-parallelism': '32'
        },
        application_args=[
            '--input', 'kafka',
            '--bootstrap-servers', 'kafka-broker:9092',
            '--group-id', 'user-activity-group',
            '--output', 'elasticsearch',
            '--es-nodes', 'es-node:9200'
        ],
        jm_conn_id='flink_jobmanager', # Optional: Explicitly define JobManager connection
        task_manager_conn_id='flink_taskmanager', # Optional: Explicitly define TaskManager connection
    )

    process_batch_flink_job = FlinkOperator(
        task_id='process_daily_reports',
        flink_conn_id='flink_cluster',
        application='/opt/flink/my-batch-processor.jar',
        job_name='daily-report-aggregation',
        parallelism=2,
        savepoint_path=None, # Optional: Restore from a savepoint
        allow_non_restored_state=False, # Optional: Prevent starting if state cannot be fully restored
        application_args=[
            '--input-path', '/data/daily_input',
            '--output-path', '/reports/daily_aggregation'
        ],
    )

    trigger_downstream_report = BashOperator(
        task_id='trigger_reporting_system',
        bash_command='echo "Flink jobs completed. Triggering report generation..." && sleep 5',
        trigger_rule='all_success',
    )

    prepare_data >> submit_realtime_flink_job >> process_batch_flink_job >> trigger_downstream_report

2. Flink as a Task within an Airflow DAG: Granular Pipeline Steps

Here, Flink handles specific data transformation stages within a broader Airflow workflow that might involve other technologies.

Example Scenario:

An ETL pipeline where Airflow extracts data from various sources, uses a Flink job to perform complex stream-based enrichment, and then loads the processed data into a data warehouse using another Airflow operator.

• Airflow tasks could handle data extraction from databases, APIs, or cloud storage.
• A FlinkOperator task would then process this data in real-time for enrichment or complex transformations.
• Subsequent Airflow tasks could load the processed data into systems like Snowflake or Redshift using their respective Airflow operators.

3. Interacting with Flink’s REST API from Airflow Tasks: Advanced Control

For more fine-grained control, Airflow tasks can directly interact with Flink’s REST API.

Use Cases:

• Dynamically submitting Flink jobs with configurations generated by upstream Airflow tasks (e.g., using XComs to pass parameters).
• Implementing custom monitoring logic beyond what the FlinkOperator provides.
• Triggering specific actions within a long-running Flink application (if the application exposes such endpoints).
• Checking the health or metrics of a running Flink job before proceeding with downstream tasks.

Mechanism:

This typically involves using the PythonOperator in Airflow and leveraging Python libraries like requests to make HTTP calls to the Flink JobManager’s REST API endpoints. You’ll need to handle authentication and API versioning.

4. Flink Triggering Airflow DAGs: Event-Driven Workflows

This pattern is less common but enables event-driven workflows where Flink’s real-time processing can trigger downstream Airflow DAGs.

Mechanism:

• Airflow REST API Listener: You can configure the Airflow REST API (ensure it’s secured) and have your Flink application make HTTP POST requests to trigger DAG runs based on specific processing outcomes or events within Flink.
• External Triggering System: Flink can write a trigger event (e.g., a message to a Kafka topic, a file to a specific location) that an Airflow sensor task (e.g., KafkaSensor, FileSensor) monitors. Upon detecting the event, the sensor task completes, allowing the downstream DAG to proceed.

Use Cases:

Could be used for triggering batch reporting or alerting workflows based on real-time insights generated by Flink.