In enterprise data engineering and applied data science, raw telemetry, transaction logs, and customer interactions originate across disparate heterogeneous operational systems. Before these disparate streams can power predictive machine learning models, executive analytical dashboards, or real-time business intelligence, they must be systematically extracted, cleaned, normalized, enriched, and loaded into centralized analytical stores. Workflow orchestration provides the programmatic backbone that coordinates these multi-stage computational pipelines, guaranteeing reliability, execution ordering, fault tolerance, and data freshness.
This guide provides an exhaustive, university-grade examination of modern pipeline orchestration using Apache Airflow. We analyze the architectural shift from legacy ETL to cloud-native ELT, explore graph-theoretic topological scheduling, examine the internal micro-architecture of the Airflow scheduler and distributed executors, derive idempotent partition engineering, and formulate failure mitigation strategies.
ETL vs. ELT: Architectural Evolution#
The paradigm of data pipeline engineering has undergone a fundamental transformation driven by the economics of cloud storage and massively parallel processing (MPP) analytical data warehouses (such as Snowflake, Google BigQuery, and Amazon Redshift).
+---------------------------------------------------------------------------------------------------+
| ETL vs. ELT Architectural Topology |
+---------------------------------------------------------------------------------------------------+
| 1. Traditional ETL (Extract -> Transform -> Load) |
| +---------------+ +---------------------+ +---------------------+ |
| | Source OLTP | ---> | Dedicated ETL Server| ---> | Analytical DB | |
| | Databases / S3| | (Compute Bottleneck)| | (Pre-computed Mart) | |
| +---------------+ +---------------------+ +---------------------+ |
| Heavy compute in-flight |
| |
| 2. Modern ELT (Extract -> Load -> Transform) |
| +---------------+ +---------------------+ +---------------------+ |
| | Source OLTP | ---> | Raw Data Lake | ---> | Cloud MPP Warehouse | |
| | Databases / S3| | (S3 / GCS / Parquet)| | (dbt / SQL Compute) | |
| +---------------+ +---------------------+ +---------------------+ |
| Zero compute load Distributed SQL Pushdown |
+---------------------------------------------------------------------------------------------------+Traditional ETL (Extract, Transform, Load)#
In legacy on-premises data architectures, storage and compute were severely constrained and tightly coupled. Operational databases extracted raw data, which was streamed into dedicated transformation compute engines (e.g., Informatica, IBM InfoSphere, or bespoke Python/Java worker clusters).
- Raw data was parsed, validated, scrubbed, and aggregated in-memory prior to loading.
- Only the final, sanitized analytical aggregates were loaded into the destination data warehouse.
- Disadvantages: Compute bottlenecks on transformation nodes; if an error was discovered in business transformation logic months later, historical raw data was already discarded and could never be reprocessed.
Modern ELT (Extract, Load, Transform)#
Cloud object storage (AWS S3, Google Cloud Storage, Azure ADLS) offers virtually infinite capacity at negligible commodity cost. Modern architectures decouple extraction from transformation:
- Extract & Load: Raw, untransformed data is ingested directly from sources and loaded immediately into an immutable raw data lake (Bronze tier).
- Transform: Transformations are executed in-situ inside modern analytical databases or distributed query engines (Apache Spark, Trino, ClickHouse, dbt) leveraging columnar storage and vectorized execution kernels.
- Advantages: High ingest throughput; complete historical auditability; transformations can be iterated, modified, and backfilled retrospectively without re-querying operational source databases.
Graph Theory Foundations: Directed Acyclic Graphs (DAGs)#
Data pipelines are naturally represented as Directed Acyclic Graphs (DAGs):
G = (V, E)where:
- Vertices (V): Represent discrete, atomic units of computation (tasks, extraction jobs, SQL queries, model inference scripts).
- Directed Edges (E): Represent execution dependencies and precedence constraints. A directed edge
(u, v) ∈ Eindicates that taskvcannot commence execution until taskuhas completed successfully (u → v).
The Acyclic Constraint#
A workflow graph must be strictly acyclic: there exists no directed path starting and ending at the same vertex (v_0 → v_1 → ... → v_k → v_0). If a cycle existed, the orchestration scheduler would enter an infinite deadlock condition where task u waits for task v, which transitively waits for task u.
Topological Sorting: Kahn's Algorithm#
To determine an executable sequence of tasks that honors all dependency constraints, the workflow engine computes a topological ordering using Kahn's algorithm:
- Compute the in-degree
in_deg(v)for each vertexv ∈ V(the number of incoming edges pointing tov). - Initialize a queue Q containing all vertices with in-degree 0 (
in_deg(v) = 0), representing tasks with zero unmet prerequisites. - While Q is non-empty:
* Dequeue vertex u and append it to the topological ordering list L.
* For each outgoing neighbor w of u (edge (u, w)):
* Decrement in_deg(w) by 1.
* If in_deg(w) == 0, enqueue w into Q.
- If the number of vertices in
Lis less than|V|, the graph contains at least one directed cycle, and the workflow is rejected as mathematically invalid.
+---------------------------------------------------------------------------------------+
| Pipeline Dependency Graph |
+---------------------------------------------------------------------------------------+
| |
| [ Extract Postgres ] ----\ |
| +---> [ Staging Cleanse ] ---> [ Train ML Model ] |
| [ Extract Stripe API ] --/ | | |
| v v |
| [ Aggregate Daily ] [ Deploy Model Canary ] |
| |
+---------------------------------------------------------------------------------------+Apache Airflow: Architecture & Internal Mechanics#
Apache Airflow (created at Airbnb in 2014 and graduated to top-level Apache project in 2019) is a platform to author, schedule, and monitor workflows programmatically as code in Python.
+---------------------------------------------------------------------------------------+
| Apache Airflow Architecture |
+---------------------------------------------------------------------------------------+
| |
| Webserver (UI) <========================================+ |
| ^ | |
| | Queries | |
| v v |
| [ Metadata Database ] <---> [ Airflow Scheduler ] <---> [ Airflow Triggerer ] |
| (PostgreSQL / MySQL) (Heartbeat / Polling Loop) (Async Deferrable Coroutines)
| ^ | |
| | v Submits Tasks |
| | [ Executor Interface ] |
| | (Celery / Kubernetes) |
| | | |
| | v |
| +----------------------- [ Worker Pool ] |
| (Executes Tasks) |
| |
+---------------------------------------------------------------------------------------+Core Architectural Components#
- Airflow Scheduler: The operational heart of the system. It runs an infinite loop consisting of three phases:
DAG File Processing*: Continuously parses Python files in the DAG directory, extracting task dependencies and cron schedules.
Task State Evaluation*: Queries the metadata database to identify task instances whose upstream dependencies have succeeded and whose scheduled data intervals have elapsed, transitioning their state from none to scheduled to queued.
Executor Dispatch*: Dispatches queued tasks to the configured executor for physical execution.
- Metadata Database: A relational database (typically PostgreSQL) storing the persistent state of all DAG runs, task instances, connection credentials, user permissions, and variables.
- Webserver: A Flask and Gunicorn web application providing a rich user interface to visualize DAG runs, inspect logs, trigger manual retries, and manage system connections.
- Triggerer: An event-driven, asynchronous coroutine runner (built on Python
asyncio). It monitors deferrable operators (such as waiting for an external Spark job, Databricks run, or S3 file sensor), freeing up worker execution slots and allowing a single triggerer process to handle thousands of concurrent deferred tasks. - Workers: Compute processes or containers that physically execute the task code.
Distributed Execution Engines#
Airflow decouples scheduling logic from task execution through pluggable Executors:
- SequentialExecutor: Executes one task at a time sequentially using SQLite. Suitable exclusively for local debugging.
- LocalExecutor: Spawns multiple worker processes locally using Python
multiprocessing. Limited by the CPU and memory of a single virtual machine. - CeleryExecutor: Uses Celery distributed task queues with Redis or RabbitMQ as the message broker. Multiple worker VMs pull tasks from queues concurrently, enabling horizontal scaling across a cluster of permanent nodes.
- KubernetesExecutor: Dynamically spawns an isolated Kubernetes Pod for every individual task instance. When the task completes, the pod terminates and frees cluster resources. Provides total dependency isolation (different tasks can run different Docker images and Python virtual environments) and zero idle worker infrastructure costs.
Authoring Workflows: The TaskFlow API#
Modern Airflow (Airflow 2.0+) introduced the TaskFlow API, utilizing Python decorators (@dag, @task) to drastically streamline DAG authoring and automate data passing via XCom (Cross-Communication).
from datetime import datetime, timedelta
from airflow.decorators import dag, task
import requests
import json
default_args = {
'owner': 'data-engineering',
'depends_on_past': False,
'email_on_failure': True,
'retries': 3,
'retry_delay': timedelta(minutes=5),
'retry_exponential_backoff': True,
'max_retry_delay': timedelta(minutes=30),
}
@dag(
dag_id='user_analytics_daily_pipeline',
default_args=default_args,
description='Daily ingestion, transformation, and feature store generation',
schedule_interval='@daily',
start_date=datetime(2026, 1, 1),
catchup=False,
max_active_runs=1,
tags=['production', 'analytics']
)
def user_analytics_pipeline():
@task(task_id='extract_raw_transactions')
def extract_transactions() -> list:
# Simulated extraction from external payment API
response = requests.get('https://api.devthesis.com/internal/transactions?date={{ ds }}')
data = response.json()
return data['records']
@task(task_id='clean_and_aggregate')
def transform_records(records: list) -> dict:
total_volume = sum(float(r['amount']) for r in records if r['status'] == 'CONFIRMED')
active_users = len(set(r['user_id'] for r in records))
return {
'confirmed_volume': total_volume,
'unique_transacting_users': active_users,
'mean_basket_size': total_volume / max(active_users, 1)
}
@task(task_id='load_to_data_warehouse')
def load_warehouse(metrics: dict):
# Database upsert logic into Snowflake / BigQuery
print(f"Executing idempotent upsert: {json.dumps(metrics)}")
# Define DAG execution dependencies via functional return values
raw_data = extract_transactions()
transformed_metrics = transform_records(raw_data)
load_warehouse(transformed_metrics)
# Instantiate the DAG
user_analytics_pipeline()XCom Mechanics and Architectural Boundaries#
XCom (Cross-Communication) enables tasks within a DAG run to exchange state. Under the hood, Airflow serializes task return values (typically as JSON) and writes them directly into the xcom table in the metadata database.
- Critical Architectural Constraint: The metadata database is an operational transactional store, not a high-throughput data lake. Storing large DataFrames, raw CSV payloads, or model weights in XCom exhausts database memory and locks connections.
- Best Practice: Use XCom exclusively for tiny metadata coordinates: S3 URIs (
s3://bucket/data/2026-10-30/part-001.parquet), row counts, execution IDs, or schema version tokens.
Airflow Temporal Mechanics: Execution Dates vs. Logical Dates#
One of the most frequent sources of confusion for data engineers is Airflow's scheduling interval mechanics.
The Data Interval Window#
In Airflow, a DAG does not execute at the beginning of its time window; it executes at the end of the interval, once all data for that interval has physically arrived.
For a DAG scheduled @daily:
- Data Interval Start:
2026-10-30 00:00:00 UTC - Data Interval End:
2026-10-30 23:59:59 UTC - Execution Trigger Time:
2026-10-31 00:00:00 UTC(The instant the day concludes).
In Airflow 2.2+, the historical execution_date macro was superseded by logical_date and data_interval_start:
{{ data_interval_start }}: Marks the beginning of the period covered by the pipeline.{{ data_interval_end }}: Marks the end of the period.{{ ds }}: YYYY-MM-DD formatted string of the data interval start.
Backfilling & Catchup#
When a new DAG is deployed with start_date = datetime(2026, 1, 1) and current date is 2026-06-01:
- `catchup = True`: The scheduler immediately generates and schedules DAG runs for every single daily interval between Jan 1 and June 1 (150+ concurrent DAG runs), rapidly backfilling historical data.
- `catchup = False`: The scheduler skips all historical intervals and schedules only the single most recent data interval.
Pipeline Idempotency & Fault Tolerance#
A data pipeline is idempotent if running it multiple times with the identical input parameters produces the exact same end state without duplicate records, data corruption, or unintended side effects:
f(f(x)) = f(x)In distributed environments, tasks fail frequently due to network partitions, spot instance evictions, out-of-memory (OOM) errors, or downstream API rate limits. Without idempotency, retrying a failed task that crashed halfway through writing records will insert duplicate rows into analytical tables.
+---------------------------------------------------------------------------------------+
| Idempotent Storage Patterns |
+---------------------------------------------------------------------------------------+
| Anti-Pattern: Imperative Append |
| INSERT INTO orders SELECT * FROM staging_orders; |
| ===> Fails halfway -> Retried -> 10,000 duplicate orders created! |
| |
| Pattern 1: Partition Overwrite (Atomic File Swap) |
| INSERT OVERWRITE TABLE orders PARTITION (order_date = '2026-10-30') |
| SELECT * FROM staging_orders; |
| ===> Idempotent: 10 re-runs produce identical partition state. |
| |
| Pattern 2: Merge / Upsert via Primary Key Deduplication |
| MERGE INTO target_table t USING source_table s ON t.id = s.id |
| WHEN MATCHED THEN UPDATE SET t.val = s.val |
| WHEN NOT MATCHED THEN INSERT (id, val) VALUES (s.id, s.val); |
+---------------------------------------------------------------------------------------+Exponential Backoff Retry Math#
When external APIs or database connections experience transient outages, immediate retries exacerbate network congestion. Airflow implements Exponential Backoff:
Delay(k) = min( T_base · 2^(k - 1), T_max )where:
kis the retry attempt index (k = 1, 2, ..., N_retries).T_baseis the initial retry delay.T_maxis the ceiling cap to prevent astronomical wait times.
Advanced Sensor Patterns: Poke vs. Reschedule Mode#
Sensors are specialized Airflow operators that continuously evaluate an external condition (such as the arrival of an S3 file key, the completion of a partition in Hive/Snowflake, or an HTTP 200 health check) before releasing downstream tasks.
The Worker Starvation Anti-Pattern: Poke Mode#
In default mode='poke', a sensor executes a tight polling loop:
- The sensor task acquires an execution slot on an Airflow worker process.
- It evaluates the condition (
poke()). If false, it sleeps forpoke_intervalseconds (e.g., 60 seconds) while holding the worker slot occupied. - If an enterprise pipeline has 50 sensor tasks waiting for partner files that arrive hours later, all 50 worker concurrency slots remain permanently locked in sleep states. Legitimate transformation tasks are starved of compute, bringing the entire Airflow cluster to a standstill.
The Scalable Alternative: Reschedule Mode & Deferrable Operators#
Modern Airflow architectures mandate two scalable sensor patterns:
- `mode='reschedule'`:
When poke() evaluates to false, the sensor immediately frees up the worker slot, enters the up_for_reschedule state, and writes its wake-up timestamp to the metadata database. The worker is instantly returned to the pool to execute heavy transformation tasks. When poke_interval elapses, the scheduler queues the sensor again.
- Deferrable Operators & The Airflow Triggerer:
Deferrable operators utilize Python's asynchronous coroutine loop (asyncio). Instead of taking up worker threads, thousands of waiting tasks yield control to a single lightweight airflow-triggerer daemon. When the external event occurs (e.g., an S3 object notification fires), the triggerer notifies the scheduler to resume the task on an active worker.
+---------------------------------------------------------------------------------------+
| Sensor Concurrency Execution Models |
+---------------------------------------------------------------------------------------+
| Model 1: Poke Mode (Slot Hogging Anti-Pattern) |
| [ Worker Slot 1 ] ===> [ Sensor Sleep 60s ] ===> [ Slot Permanently Blocked ] |
| |
| Model 2: Reschedule Mode (Worker Slot Releasing) |
| [ Worker Slot 1 ] ===> [ Poke False ] ===> [ Release Slot ] ===> [ Worker Free ] |
| | |
| v |
| [ Metadata DB holds wake-up timer ] |
| |
| Model 3: Deferrable Operator (Event-Driven Triggerer) |
| [ Worker Slot ] ---> [ Yield to Triggerer ] ---> [ 1 Triggerer holds 10,000 tasks ]|
+---------------------------------------------------------------------------------------+Dynamic DAG Generation & TaskGroup Architecture#
In large-scale enterprise environments, hard-coding separate static Python DAG files for hundreds of client tables or database replication jobs creates massive code duplication and maintenance friction.
Dynamic DAG Factory Pattern#
Airflow parses the global namespace of Python scripts in the DAG directory. By iterating over a metadata configuration dictionary (or external YAML/JSON configuration files), a single script can dynamically instantiate dozens of independent DAGs at parse time:
# Dynamic DAG Generation from Configuration Metadata
TABLES_TO_REPLICATE = [
{'name': 'customers', 'schedule': '@daily', 'retries': 3},
{'name': 'orders', 'schedule': '0 * * * *', 'retries': 5},
{'name': 'inventory', 'schedule': '*/15 * * * *', 'retries': 2},
]
for table in TABLES_TO_REPLICATE:
dag_id = f"replicate_{table['name']}_to_warehouse"
@dag(
dag_id=dag_id,
schedule_interval=table['schedule'],
start_date=datetime(2026, 1, 1),
catchup=False,
default_args={'retries': table['retries']}
)
def dynamic_replication_dag():
@task
def extract_delta():
return f"Extracted delta for {table['name']}"
@task
def load_warehouse(msg):
print(f"Loading {msg}")
load_warehouse(extract_delta())
# Instantiate into global scope for Airflow scheduler discovery
dynamic_replication_dag()TaskGroups vs. Legacy SubDAGs#
In early Airflow versions, complex modular pipelines relied on SubDAGOperator, which had severe architectural flaws (SubDAGs ran on isolated secondary schedulers, leading to deadlocks and database connection starvation).
Airflow 2.0 replaced SubDAGs with `TaskGroup`: a purely UI-level and organizational abstraction that groups related tasks into collapsible visual modules without introducing separate scheduling loops or concurrency deadlocks.
Checkpoint Practice Problems#
Problem 1 (DAG Topological Sort via Kahn's Algorithm): Consider a data pipeline with 6 tasks:
V = { A, B, C, D, E, F }
The dependency edges are:
E = { (A, B), (A, C), (B, D), (C, D), (C, E), (D, F), (E, F) }
- Calculate the initial in-degree for every vertex.
- Step through Kahn's algorithm step-by-step, showing the contents of Queue Q, the vertex processed, and the updating in-degrees.
- State the resulting topological execution sequence L.
- If an additional edge
(F, C)were accidentally added, explain how Kahn's algorithm detects the cycle.
Solution:
- Initial in-degrees:
* A: 0 (No incoming edges)
* B: 1 (from A)
* C: 1 (from A)
* D: 2 (from B and C)
* E: 1 (from C)
* F: 2 (from D and E)
- Trace of Kahn's Algorithm:
* Step 1: Initial Q = [ A ].
* Dequeue A. L = [ A ].
* Remove edge (A, B): in-degree(B) becomes 0 -> Enqueue B.
* Remove edge (A, C): in-degree(C) becomes 0 -> Enqueue C.
* Q = [ B, C ].
* Step 2: Dequeue B. L = [ A, B ].
* Remove edge (B, D): in-degree(D) = 2 - 1 = 1.
* Q = [ C ].
* Step 3: Dequeue C. L = [ A, B, C ].
* Remove edge (C, D): in-degree(D) = 1 - 1 = 0 -> Enqueue D.
* Remove edge (C, E): in-degree(E) = 1 - 1 = 0 -> Enqueue E.
* Q = [ D, E ].
* Step 4: Dequeue D. L = [ A, B, C, D ].
* Remove edge (D, F): in-degree(F) = 2 - 1 = 1.
* Q = [ E ].
* Step 5: Dequeue E. L = [ A, B, C, D, E ].
* Remove edge (E, F): in-degree(F) = 1 - 1 = 0 -> Enqueue F.
* Q = [ F ].
* Step 6: Dequeue F. L = [ A, B, C, D, E, F ].
* Q is now empty.
- Resulting valid topological sort:
L = [ A, B, C, D, E, F ] (or [ A, C, B, E, D, F ]). All dependencies are strictly preserved.
- Cycle detection with edge
(F, C):
If (F, C) exists, the in-degree of C is initially 2 (from A and F).
* After processing A, in-degree of C decreases from 2 to 1 (not 0).
* C is never added to Q.
* Tasks D, E, and F depend on C and never have their in-degrees reduced to 0.
* The algorithm halts with L = [ A, B ].
* Since |L| = 2 < |V| = 6, the algorithm flags that 4 nodes remain unprocessed, proving the existence of a directed cycle (C → D/E → F → C).
Problem 2 (Exponential Backoff Duration Calculation): An Airflow task configured to ingest external API records fails repeatedly. The task parameters are:
- Maximum retries:
N_retries = 4 - Base retry delay:
T_base = 3 minutes - Maximum retry delay cap:
T_max = 20 minutes - Each execution attempt runs for
T_run = 2 minutesbefore encountering an HTTP 504 Gateway Timeout.
Compute:
- The delay duration for each of the 4 retry attempts.
- The total cumulative elapsed time from the start of the initial attempt until the task finally triggers a fatal failure alert.
Solution:
- Compute retry delays:
* Attempt 1 (Initial run): Runs 2 min -> Fails.
* Retry 1: Delay = min(3 · 2^0, 20) = min(3, 20) = 3 minutes.
* Retry 2: Delay = min(3 · 2^1, 20) = min(6, 20) = 6 minutes.
* Retry 3: Delay = min(3 · 2^2, 20) = min(12, 20) = 12 minutes.
* Retry 4: Delay = min(3 · 2^3, 20) = min(24, 20) = 20 minutes (Capped at T_max).
- Cumulative elapsed time:
* Execution time across 5 attempts (1 initial + 4 retries):
5 attempts · 2 minutes/attempt = 10 minutes.
* Wait time across 4 backoff intervals:
3 + 6 + 12 + 20 = 41 minutes.
Total elapsed time = 10 + 41 = 51 minutes*.
The task alerts on-call engineers 51 minutes after the initial attempt.
Problem 3 (Airflow Data Interval Alignment & Cron Scheduling): A daily aggregation pipeline has the following configuration:
schedule_interval = '0 3 * * *'(Runs at 03:00 AM UTC every day).start_date = datetime(2026, 3, 10, 0, 0, 0)
For the third DAG run generated by this pipeline:
- What is the
data_interval_starttimestamp? - What is the
data_interval_endtimestamp? - At what exact calendar timestamp does the Airflow Scheduler first trigger this DAG run?
Solution:
- Schedule sequence:
* Run 1:
* Data interval start: 2026-03-10 03:00:00 UTC
* Data interval end: 2026-03-11 03:00:00 UTC
* Trigger timestamp: 2026-03-11 03:00:00 UTC
* Run 2:
* Data interval start: 2026-03-11 03:00:00 UTC
* Data interval end: 2026-03-12 03:00:00 UTC
* Trigger timestamp: 2026-03-12 03:00:00 UTC
* Run 3:
* Data interval start: 2026-03-12 03:00:00 UTC
* Data interval end: 2026-03-13 03:00:00 UTC
* Trigger timestamp: 2026-03-13 03:00:00 UTC
- Answers:
`data_interval_start` = 2026-03-12 03:00:00 UTC*
`data_interval_end` = 2026-03-13 03:00:00 UTC*
Trigger timestamp = 2026-03-13 03:00:00 UTC*
Problem 4 (Idempotent Partition Deduplication Sizing): A streaming ingest pipeline extracts transaction logs into a partitioned data lake table.
- Average ingestion rate:
R = 5,000 records per minute. - Daily duration: 24 hours (1,440 minutes).
- Due to an upstream network retry storm, 8% of all raw records emitted are exact duplicates (sharing identical
transaction_id). - Each record has an uncompressed size of
200 bytes.
Compute:
- The total count of raw records ingested in a 24-hour partition.
- The count of duplicate records and unique valid records.
- The storage savings (in megabytes) achieved by running an idempotent deduplication task (
ROW_NUMBER() OVER (PARTITION BY transaction_id ORDER BY event_time DESC) = 1) prior to persisting the final analytical partition (1 MB = 1,000,000 bytes).
Solution:
- Total raw records ingested:
```
Total Raw = 5,000 records/min · 1,440 min = 7,200,000 records
```
- Deduplication breakdown:
```
Duplicate Records = 7,200,000 · 0.08 = 576,000 records
Unique Records = 7,200,000 - 576,000 = 6,624,000 records
```
- Storage calculations:
* Raw partition size:
7,200,000 · 200 bytes = 1,440,000,000 bytes = 1,440.0 MB
* Deduplicated partition size:
6,624,000 · 200 bytes = 1,324,800,000 bytes = 1,324.8 MB
* Storage savings:
1,440.0 MB - 1,324.8 MB = 115.2 MB
Running the idempotent deduplication eliminates 576,000 redundant records and reduces daily partition footprint by 115.2 MB.