How to Set Up Event-Driven Data Pipelines Without Airflow

A cron job that fires at 6am doesn't know whether the source data actually landed at 5:58 or 6:47. It just runs. If the data's late, the pipeline produces stale results. If the data arrived early, you wasted 6 hours of latency waiting for the clock to tick over.

This is the fundamental tension with schedule-driven orchestration. Your pipelines operate on a fixed cadence, but the data they process doesn't. New files land when an upstream system finishes its export. A Snowflake table updates when a CDC stream catches up. Kafka messages arrive whenever they arrive. Event-driven pipelines eliminate the gap by reacting to these changes directly — no scheduler polling empty tables, no cron jobs running against yesterday's data.

Airflow can handle event-driven workloads. Its sensor operators poll for conditions — a file in S3, a partition in Hive, an external DAG completing. But sensors hold a worker slot while they wait, which means you're either wasting compute or tuning poke_interval and mode across dozens of sensors. That's real work, on top of the scheduler, metadata database, web server, and worker fleet you're already running.

There's a simpler model.

Triggers, not sensors

The event-driven alternative inverts the polling model. Instead of a running process that checks "has the thing happened yet?" every N seconds, you register a trigger: when the thing happens, run this. No idle workers. No poke intervals. No sensor timeout edge cases.

In Fastero, triggers are the entry point for every workflow. You define what to listen for — a Snowflake stream change, a Kafka message, a file landing in a stage, a cron tick — and attach a sequence of actions. The trigger fires, the workflow runs, and nothing sits idle between events.

Here's what that looks like across four real scenarios.

Scenario 1: New data in a Snowflake stream

Snowflake streams track row-level changes (inserts, updates, deletes) on a table. Every time the source table changes, the stream accumulates the delta. Fastero's Snowflake stream trigger checks whether a stream has unconsumed changes and fires your workflow when it does.

trigger:
  type: snowflake_stream
  connection: warehouse-prod
  stream: RAW.PUBLIC.ORDERS_STREAM
  check_interval: 60s
 
steps:
  - name: merge_new_orders
    action: run_sql
    sql: |
      MERGE INTO ANALYTICS.PUBLIC.FACT_ORDERS t
      USING RAW.PUBLIC.ORDERS_STREAM s
      ON t.order_id = s.order_id
      WHEN MATCHED THEN UPDATE SET
        t.status = s.status,
        t.updated_at = s.updated_at
      WHEN NOT MATCHED THEN INSERT
        (order_id, customer_id, amount, status, created_at, updated_at)
        VALUES (s.order_id, s.customer_id, s.amount, s.status, s.created_at, s.updated_at);
 
  - name: refresh_revenue_dashboard
    action: refresh_dashboard
    dashboard: daily-revenue

The stream trigger checks every 60 seconds. When there are unconsumed rows, it runs the merge and refreshes the dashboard. When there aren't, nothing happens — no worker sitting idle, no wasted compute.

Compare this to the Airflow equivalent: a SnowflakeSensor holding a worker slot, a SnowflakeOperator for the merge, a PythonOperator for the dashboard refresh, all wired into a DAG with its own schedule, retries, and timeout config. That's five files and two abstractions for what's essentially one trigger and two steps.

Scenario 2: Kafka topic triggers a data quality check

When a Kafka consumer group processes messages into your warehouse, you often want a downstream check — are there nulls in required fields, did the row count spike unexpectedly, is the latest timestamp within the expected window.

trigger:
  type: kafka
  connection: kafka-prod
  topic: payments.processed
  consumer_group: fastero-dq-checks
  batch_size: 100
 
steps:
  - name: check_nulls
    action: run_sql
    sql: |
      SELECT COUNT(*) AS null_count
      FROM analytics.payments
      WHERE payment_id IS NULL
        OR amount IS NULL
        OR processed_at > CURRENT_TIMESTAMP - INTERVAL '5 minutes';
    store_result: null_check
 
  - name: alert_on_nulls
    action: send_notification
    condition: "{{ null_check.null_count > 0 }}"
    channel: slack
    destination: "#data-alerts"
    message: "⚠️ {{ null_check.null_count }} null records in payments table"

Every batch of 100 messages triggers the quality check. The condition on the alert step means Slack only gets pinged when something's actually wrong — not on every batch.

This pattern replaces what would be a KafkaSensor (or a custom KafkaConsumerOperator) plus a SQLCheckOperator plus a SlackAPIPostOperator in Airflow, each requiring its own connection configuration and error handling. For more on building real-time dashboards on top of Kafka data, see our Kafka dashboard guide.

Scenario 3: File lands in a Snowflake stage

External data partners love dropping CSV files on a schedule that isn't actually a schedule. "Daily by 8am" means sometimes 7:15, sometimes 10:30, occasionally not at all. A cron job at 9am misses the late ones and wastes a run when there's nothing there.

Snowflake's external stages can be monitored for new files. The stage trigger fires when a new file appears:

trigger:
  type: snowflake_stage
  connection: warehouse-prod
  stage: "@RAW.PUBLIC.PARTNER_UPLOADS"
  file_pattern: "orders_*.csv.gz"
 
steps:
  - name: load_file
    action: run_sql
    sql: |
      COPY INTO RAW.PUBLIC.PARTNER_ORDERS
      FROM @RAW.PUBLIC.PARTNER_UPLOADS
      FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1 FIELD_OPTIONALLY_ENCLOSED_BY = '"')
      PATTERN = 'orders_.*csv.gz'
      ON_ERROR = CONTINUE;
 
  - name: validate_row_count
    action: run_sql
    sql: |
      SELECT COUNT(*) AS loaded
      FROM RAW.PUBLIC.PARTNER_ORDERS
      WHERE _loaded_at > CURRENT_TIMESTAMP - INTERVAL '10 minutes';
    store_result: load_result
 
  - name: notify_team
    action: send_notification
    channel: slack
    destination: "#data-ops"
    message: "Loaded {{ load_result.loaded }} partner order rows"

File arrives at 7:15? Pipeline runs at 7:15. File arrives at 10:30? Pipeline runs at 10:30. File doesn't arrive at all? Nothing runs, and you can set up a separate scheduled alert to catch the absence.

Scenario 4: Multi-step workflow with conditional branching

Real pipelines aren't always linear. Sometimes step 3 depends on the result of step 2. Sometimes you want to run different paths based on data characteristics. Fastero workflows support conditions at each step — not a full DAG, but enough branching to handle the patterns that actually show up in practice.

trigger:
  type: cron
  schedule: "0 */4 * * *"  # every 4 hours
 
steps:
  - name: check_data_freshness
    action: run_sql
    sql: |
      SELECT
        DATEDIFF('minute', MAX(updated_at), CURRENT_TIMESTAMP) AS minutes_stale
      FROM analytics.fact_orders;
    store_result: freshness
 
  - name: run_full_rebuild
    action: run_sql
    condition: "{{ freshness.minutes_stale > 240 }}"
    sql: |
      TRUNCATE TABLE analytics.fact_orders;
      INSERT INTO analytics.fact_orders
      SELECT * FROM raw.orders_enriched;
 
  - name: run_incremental
    action: run_sql
    condition: "{{ freshness.minutes_stale <= 240 }}"
    sql: |
      INSERT INTO analytics.fact_orders
      SELECT * FROM raw.orders_enriched
      WHERE updated_at > (SELECT MAX(updated_at) FROM analytics.fact_orders);
 
  - name: update_dashboards
    action: refresh_dashboard
    dashboard: executive-metrics

This one uses a cron trigger — event-driven doesn't mean you never use schedules. The distinction is that the workflow adapts to data state rather than blindly re-running regardless of what changed. More than 4 hours stale? Full rebuild. Otherwise, incremental.

When to stay with Airflow

Event-driven pipelines aren't universally better. They're better for a specific shape of work — and it happens to be the shape most teams under 50 jobs actually have.

Stick with Airflow (or Dagster) when you have genuine dependency trees — ten transforms that must execute in a specific order, with fan-out and fan-in, with partial retries on individual branches. DAG-based orchestrators model this well. Event-driven triggers model it awkwardly. Also stick with Airflow when you need a single control plane across hundreds of jobs.

But if your "DAG" is really three steps running sequentially, that's not a dependency graph. That's a list. And a workflow with three steps and a trigger is a much simpler way to run a list than a DAG definition with imports, default_args, operator instantiation, and >> chains.

The migration path

If you're running Airflow now, the migration is straightforward. Identify DAGs that are a single chain of tasks with a schedule or sensor at the top — each one maps to a trigger + workflow with the same number of steps. Keep the DAGs with real fan-out/fan-in dependencies. Move the linear ones. Fastero triggers don't need to replace your entire orchestration layer. They just take the simple jobs off the pile so your Airflow instance handles fewer things and needs less capacity.

For teams running scheduled Python jobs that outgrew cron but don't need full orchestration, the workflow model slots in the same way — more structure than a script on a timer, less overhead than a platform.

Try Fastero free — set up event-driven pipelines with Snowflake, Kafka, and cron triggers in minutes, not weeks. No credit card required.