vault/Career/Microlise/Competencies/Synchronous vs Asynchronous Dataflow Design.md

# Competency Evidence: Synchronous vs Asynchronous Dataflow Design

**Competency:** Discerning between synchronous and asynchronous programming to improve dataflow choices within an application.

**Author:** Zaine Qayyum  
**Team:** Osmosis  
**Date:** June 2026  
**Primary artefact:** Dora Orchestrator Airflow pipeline (`DoraOrchestrator` DAG)

---

## 1. Competency Statement

### Value of achieving this competency

Being able to discern between using synchronous and asynchronous programming to improve dataflow choices within your application.

### Scenario mapping

The competency scenario describes fetching data from multiple sources to populate a dashboard, where some data can be displayed immediately and some depends on other data being available first. The Dora Orchestrator pipeline is a direct realisation of this scenario:

- **Independent data domains** (builds, deployments, releases/environments) can be collected in parallel as soon as staging is prepared.
    
- **Dependent data domains** (repositories → pull requests → commits; work items and threads after pull requests) must wait for upstream parquet outputs before they can run.
    
- **A merge barrier** ensures all staging data is complete before the datamart is updated for PowerBI consumption.
    

### My role

I was the primary author and maintainer of the Dora Orchestrator pipeline, including the orchestrator DAG, all collection scripts, unit tests, and the pull requests cited in this document.

---

## 2. Problem Context

### Business need

The Osmosis team needed to surface DORA (DevOps Research and Assessment) metrics via PowerBI dashboards. Metrics are derived from multiple Azure DevOps (AZDO) and SonarQube API domains: teams, systems, repositories, pull requests, commits, work items, threads, builds, deployments, releases, and environments.

### Original approach

Data was queried directly from AZDO APIs inside PowerBI. This was brittle, tightly coupled, and effectively serial — each query blocked the next, with no orchestration layer to manage dependencies or parallelise independent fetches.

### New approach

A nightly Airflow orchestrator (`DoraOrchestrator`) collects data from AZDO and SonarQube APIs, writes to staging tables in `DoraDatabase_DEV`, executes merge stored procedures, and exposes the result to PowerBI via the [Dora DataMart](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/DataMart.Academy.DoraMetrics).

### Performance and dataflow constraints

1. **Many independent API domains** — builds, deployments, and releases do not depend on repository metadata and can run concurrently.
    
2. **Strict cross-domain dependencies** — commits require pull request IDs; pull requests require repository IDs; repositories require team/system enrichment.
    
3. **Large payloads** — tens of thousands of records per domain; inter-task data transfer must not use Airflow's metadata database.
    
4. **Operational reliability** — daily 1am schedule, Slack failure alerts, and 80%+ unit test coverage across all scripts.
    

---

## 3. Architecture Overview

The orchestrator is defined in `Dags/DAG_Osmosis_DoraOrchestrator.py`. Collection logic lives in `Dags/OSMOSIS_DoraMetrics_Scripts/`.

### DAG dependency graph

```mermaid
flowchart TD
    setup[clear_create_directory_start]
    truncate[truncate_staging_tables]
    builds[builds TaskGroup]
    systemTeams[system_teams]
    repositories[repositories]
    pullRequests[pull_requests]
    commits[commits]
    workItem[work_item]
    threads[threads]
    deployments[deployments TaskGroup]
    releases[releases_and_environments TaskGroup]
    merge[merge_staging_to_data_tables]
    cleanup[clear_create_directory_end]

    setup --> truncate
    truncate --> builds
    truncate --> systemTeams
    truncate --> deployments
    truncate --> releases
    systemTeams --> repositories --> pullRequests --> commits --> merge
    pullRequests --> workItem --> threads --> merge
    builds --> merge
    deployments --> merge
    releases --> merge
    merge --> cleanup
```

### Orchestrator dependency declarations

The parallel and serial structure is encoded explicitly in the DAG:

```310:316:Dags/DAG_Osmosis_DoraOrchestrator.py
    clear_create_directory_start >> truncate_staging_tables_task
    truncate_staging_tables_task >> builds  >> merge_staging_to_data_tables_task
    truncate_staging_tables_task >> system_teams >> repositories >> pull_requests >> commits >> merge_staging_to_data_tables_task
    pull_requests >> work_item >> threads >> merge_staging_to_data_tables_task
    truncate_staging_tables_task >> deployments >> merge_staging_to_data_tables_task
    truncate_staging_tables_task >> releases_and_environments >> merge_staging_to_data_tables_task
    merge_staging_to_data_tables_task >> clear_create_directory_end
```

Airflow's scheduler uses this graph to run independent branches on separate workers concurrently, while respecting serial constraints within each branch.

---

## 4. Design Decisions — Parallel vs Serial

|   |   |   |
|---|---|---|
|Dataflow path|Execution model|Rationale|
|`builds`|Parallel from truncate|Builds are fetched from a single AZDO endpoint with continuation tokens; no dependency on teams/repos parquet|
|`deployments`|Parallel from truncate|Independent AZDO release deployment API domain|
|`releases_and_environments`|Parallel from truncate; fan-out inside group|One fetch produces releases, artifacts, and environments parquet files; two insert tasks run in parallel after fetch|
|`system_teams → repositories → pull_requests → commits`|Serial chain|Each step reads parquet written by the previous step (e.g. commits reads `pull_requests.parquet`)|
|`work_item → threads`|Serial after `pull_requests`|Work items and threads are collected after the PR domain is available|
|`merge_staging_to_data_tables`|Barrier (waits for all branches)|Staging must be complete before merge procedures update production tables|
|`clear_create_directory_end`|Serial after merge|Cleanup only after all data is persisted|

### Parallel fan-out within releases

Inside the `releases_and_environments` TaskGroup, a single fetch task fans out to two independent insert tasks:

```276:278:Dags/DAG_Osmosis_DoraOrchestrator.py
        fetch_all_releases_task  >> insert_releases_staging_task 
    
        fetch_all_releases_task  >> insert_environments_staging_task 
```

This mirrors the dashboard scenario: once release data is fetched, releases and environments can be inserted concurrently without waiting on each other.

### Serial handoff via parquet files

Downstream tasks read files produced by upstream tasks. For example, commits depend on pull request data:

```19:31:Dags/OSMOSIS_DoraMetrics_Scripts/Commits.py
def fetch_all_commits(**context):
    collection_directory = parquet_setup(COLLECTION_NAME)
    os.makedirs(collection_directory, exist_ok=True)

    pull_requests_parquet_path = os.path.join(collection_directory, "pull_requests.parquet")

    pull_requests_df = pd.read_parquet(pull_requests_parquet_path)
    repository_id = 'repository.id'

    pairs_df = pull_requests_df[['pullRequestId', repository_id]].dropna()
    pairs_df['pullRequestId'] = pairs_df['pullRequestId'].astype(int)
    pairs_df[repository_id] = pairs_df[repository_id].astype(str)
    pairs = list(pairs_df.itertuples(index=False, name=None))
```

This serial constraint is intentional: commits cannot be meaningfully collected until pull requests exist.

---

## 5. Synchronous vs Asynchronous — Honest Tradeoff Analysis

This section distinguishes two layers of concurrency in the pipeline and explains why each approach was chosen.

### Layer 1: Orchestration-level concurrency (used)

**What it is:** Airflow runs independent TaskGroups on separate worker processes concurrently. This is workflow parallelism, not language-level `async`/`await`.

**Evidence:** Four branches fan out from `truncate_staging_tables_task` (builds, system chain, deployments, releases). The Airflow scheduler determines which tasks are runnable based on the dependency graph.

**Why this was the right choice:**

- Dependencies between domains are explicit and visible in the DAG graph.
    
- Each domain is an isolated failure unit with its own retry semantics and Slack alert.
    
- Process-level parallelism scales with Airflow worker pool capacity without rewriting collection scripts.
    

### Layer 2: In-task I/O (synchronous — used)

**What it is:** Within each task, HTTP and database calls are blocking and sequential. There is no `async`/`await`, `asyncio`, threading, or `concurrent.futures` usage anywhere in `OSMOSIS_DoraMetrics_Scripts/`.

**Evidence — synchronous per-repo API loop in pull requests:**

```65:68:Dags/OSMOSIS_DoraMetrics_Scripts/PullRequests.py
    for repo_id in repo_ids:
        pull_requests = fetch_pull_requests(repo_id) or []

        repo_url = f"https://azdo.microlise.com/MicroliseCollection/_apis/git/repositories/{repo_id}/"
```

Each `fetch_pull_requests` call blocks on `HttpHook.run()` until the AZDO API responds. Nested loops also make additional blocking calls per pull request for commit dates.

**Evidence — batched but sequential work item fetches:**

```45:51:Dags/OSMOSIS_DoraMetrics_Scripts/WorkItems.py
    for batch in spilt_json_into_chunks(work_item_ids, 200):
        ids_csv = ",".join(map(str, batch))
        endpoint = ENDPOINT_BASE + ids_csv

        response = get_api_information(WEBHOOK, endpoint, data=None, headers=None, api_name="Work Items")
        items = extract_json(response, "value")
        all_work_items.extend(items)
```

Batches of 200 reduce request count, but each batch is fetched sequentially.

**Why synchronous in-task I/O was chosen:**

1. **Airflow already parallelises at the DAG layer** — adding in-process async would overlap with worker-level concurrency without simplifying the dependency graph.
    
2. **Predictable API rate-limit behaviour** — sequential calls per worker are easier to reason about than concurrent requests from multiple async coroutines.
    
3. **Simpler error handling and testability** — each callable is a plain Python function with 80%+ unit test coverage across all domains.
    
4. **Bounded task scope** — each task has a clear start (read parquet / call API) and end (write parquet / insert staging), which maps cleanly to Airflow's task lifecycle.
    

### Layer 3: Language-level async (not used — consciously deferred)

In-process `asyncio` or thread-pool concurrency could reduce per-task runtime for I/O-bound loops (e.g. per-repo PR fetches, per-PR commit fetches). This was not implemented because:

- It would not remove the need for DAG-level dependency management between domains.
    
- It would complicate error handling, logging, and unit testing within each script.
    
- API throttling risk increases with concurrent in-task requests.
    
- The primary performance gain was achieved by parallelising independent domains at the orchestration layer, not by parallelising API calls within a single domain.
    

This is an example of **discernment**: recognising where concurrency adds value (independent domains) and where synchronous execution is the simpler correct choice (within a single domain's collection task).

### Mapping to C# async concepts

The suggested training covers C# `async`/`await` and `Task`. The conceptual mapping to this Python/Airflow pipeline is:

|   |   |
|---|---|
|C# concept|Dora Orchestrator equivalent|
|`await` on independent I/O|Parallel Airflow TaskGroups after truncate|
|`await` on dependent I/O|Serial TaskGroup chain with parquet handoffs|
|`Task.WhenAll`|Multiple branches converging at merge barrier|
|Blocking synchronous I/O|`HttpHook.run()` loops inside PythonOperator callables|
|Choosing async vs sync|DAG dependency design + XCom vs parquet dataflow decision (PR 24206)|

---

## 6. Key Pull Request Evidence

### PR 23806 — Merge Dora Metrics branch into Main

**Link:** [PR 23806](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/23806)

**Summary:** Initial delivery of the Dora Orchestrator pipeline. Moved from direct PowerBI API queries to a distributed Airflow + datamart architecture. Introduced the orchestrator DAG with TaskGroups for each AZDO domain, staging table population, merge procedures, Slack failure alerts, and unit tests exceeding 80% coverage.

**Dataflow relevance:** Established the foundational dependency structure and the principle that independent AZDO domains should be orchestrated rather than queried ad hoc.

**DAG structure diagram (from PR):**  
PR 23806 DAG structure

---

### PR 24206 — Changing XCom to parquets (primary evidence)

**Link:** [PR 24206](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24206)  
**Work item:** #106945

**Problem (from PR description):**

> This pull request makes changes mainly to the way data is being transferred across different `tasks`. As we are using `task groups` to orchestrate the flow of data, we originally used `xcom` to pull and push data between different tasks; so for example:
> 
> `fetch_all_builds runs -> pushes to xcom -> fetch_all_work_items_per_build then runs but pulls the data pushed to xcom`.
> 
> Problems with this: xcom is not suitable for large amounts of data, as the docs state: "... but they are only designed for small amounts of data; do not use them to pass around large values, like dataframes". When we first ran the DAG in the DEV environment, we ran into a memory issue, which was regarding xcom not being able to handle large amounts of data.

**Solution (from PR description):**

> Solution: Use `parquets` for writing and reading data that are transferred between tasks. As these are meant for dataframes and for large amounts of data, it makes sense to write to storage and not the airflow metadata database.
> 
> The parquet folder is created in the `shared_folder` through the task `clear_create_directory` which calls the script `ClearAndCreateCollectionDirectory`.

**Dataflow relevance:** This is the clearest example of discerning between two data-passing strategies. XCom is synchronous and lightweight (metadata DB) but wrong for large payloads. Parquet files on shared storage are the correct handoff mechanism for dataframe-sized datasets between tasks. All fetch tasks were updated with `do_xcom_push=False`.

**Implementation:** Shared folder path resolved via `parquet_setup()` in `Osmosis_Helper.py`; directory created by `ClearAndCreateCollectionDirectory.py` at DAG start.

**DAG structure diagram (from PR):**  
PR 24206 DAG structure

---

### PR 24333 — Remove the cyclic task

**Link:** [PR 24333](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24333)  
**Work item:** #108242

**Problem (from PR description):**

> The task `clear_create_directory` was called at the start and at the end, which caused a cycle loop. This PR removes that by creating separate tasks.

**Solution:** Split into `clear_create_directory_start` and `clear_create_directory_end`, preserving setup/cleanup behaviour while making the graph acyclic.

**Dataflow relevance:** A directed acyclic graph (DAG) is a prerequisite for correct parallel scheduling. A cyclic dependency would prevent Airflow from determining a valid execution order, breaking the parallel branches designed in PR 24206. This fix demonstrates understanding that dataflow graph structure directly affects concurrency.

---

### PR 24378 — Fixing the releases DAG

**Link:** [PR 24378](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24378)  
**Work item:** #108316

**Summary:** Refined the releases/environments TaskGroup. Removed unnecessary normalisation, corrected validation from `1-m` to `m-m` cardinality (one release can have multiple artifacts), and cleaned up redundant variables.

**Dataflow relevance:** Corrected the fan-out pattern where a single `fetch_all_releases` task produces data consumed by parallel `insert_releases_staging` and `insert_environments_staging` tasks. Accurate cardinality modelling ensures parallel insert tasks receive correctly shaped data.

---

### Additional supporting PRs

|   |   |   |
|---|---|---|
|PR|Title|Relevance|
|[24012](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24012)|Added a schedule for the dora dag at 1 am|Operationalised daily concurrent collection window|
|[26618](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/26618)|Adding a conversion to GUID from null repo guid values|Data integrity fix in repository enrichment chain|
|[29337](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/29337)|Updating the SQ portfolio links - Dora DAG|Maintained serial dependency chain for system/repo enrichment|
|[31719](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/31719)|Enriching systems with nested SQ Portfolios|Replaced hardcoded portfolio names with SonarQube API-driven nested portfolio fetch|

---

## 7. Outcomes and Impact

|   |   |
|---|---|
|Outcome|Detail|
|Parallel collection of independent domains|Builds, deployments, and releases/environments run concurrently after staging truncate, reducing wall-clock time versus a fully serial pipeline|
|DEV memory issue resolved|PR 24206 eliminated XCom as the inter-task transport for large dataframes, fixing the memory failure encountered on first DEV run|
|Correct DAG scheduling|PR 24333 removed the cyclic dependency, enabling Airflow to schedule parallel branches correctly|
|Daily reliable execution|PR 24012 added 1am cron schedule for unattended nightly collection|
|Test coverage maintained|Unit tests in `Dags/tests/run/Osmosis_DoraMetrics_Scripts_Tests/` maintained 80%+ coverage across all refactors|
|PowerBI-ready datamart|Merge procedures atomically promote staging data to production tables for dashboard consumption|

---

## 8. Reflection and Future Improvements

### What I would do differently with hindsight

The orchestration-layer parallelism delivered the largest performance gain for the least complexity. If per-task runtime becomes a bottleneck, the highest-impact candidates for in-task concurrency are:

- **PullRequests.py** — nested per-repo, per-PR API loops (I/O-bound, embarrassingly parallel across repos)
    
- **Commits.py** — per-PR commit fetches (I/O-bound, parallelisable across PR pairs)
    

Any move to in-process async or thread-pool concurrency would require:

1. AZDO API rate-limit analysis to avoid throttling
    
2. Airflow worker memory profiling under concurrent requests
    
3. Updated unit tests for concurrent error paths
    
4. Clear logging to preserve observability per domain
    

### Competency demonstration summary

This work evidences the competency not by using `async`/`await` everywhere, but by making deliberate dataflow choices at two levels:

1. **Where to parallelise** — independent AZDO domains via Airflow TaskGroup branches
    
2. **Where to serialise** — dependent domains via parquet handoffs and explicit `>>` dependencies
    
3. **Where to stay synchronous** — blocking API calls within tasks, because orchestration-layer concurrency already addresses the primary bottleneck
    
4. **Where to change transport mechanism** — XCom to parquet (PR 24206), because the synchronous metadata DB was the wrong medium for large async-safe handoffs
    

---

## 9. Appendix

### File inventory

|   |   |
|---|---|
|File|Purpose|
|`Dags/DAG_Osmosis_DoraOrchestrator.py`|Main orchestrator DAG — dependency graph, TaskGroups, merge barrier|
|`Dags/OSMOSIS_DoraMetrics_Scripts/SystemsTeams.py`|Teams and systems collection|
|`Dags/OSMOSIS_DoraMetrics_Scripts/Repositories.py`|Repository fetch and SonarQube enrichment|
|`Dags/OSMOSIS_DoraMetrics_Scripts/PullRequests.py`|Pull request collection (sync per-repo loops)|
|`Dags/OSMOSIS_DoraMetrics_Scripts/Commits.py`|Commit collection (reads PR parquet)|
|`Dags/OSMOSIS_DoraMetrics_Scripts/WorkItems.py`|Work item ID fetch and batched detail fetch|
|`Dags/OSMOSIS_DoraMetrics_Scripts/Threads.py`|Thread collection|
|`Dags/OSMOSIS_DoraMetrics_Scripts/Builds.py`|Build and build-work-item collection|
|`Dags/OSMOSIS_DoraMetrics_Scripts/Deployments.py`|Deployment collection|
|`Dags/OSMOSIS_DoraMetrics_Scripts/ReleasesAndEnvironments.py`|Release/environment fetch with parallel inserts|
|`Dags/OSMOSIS_DoraMetrics_Scripts/Osmosis_Helper.py`|Shared utilities — API helpers, parquet I/O, Slack alerts|
|`Dags/OSMOSIS_DoraMetrics_Scripts/ClearAndCreateCollectionDirectory.py`|Shared folder setup/cleanup|
|`Dags/tests/run/Osmosis_DoraMetrics_Scripts_Tests/`|Unit tests for all collection scripts|

### Pull request links

|   |   |
|---|---|
|PR|URL|
|23806|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/23806](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/23806)|
|24012|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24012](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24012)|
|24206|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24206](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24206)|
|24333|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24333](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24333)|
|24378|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24378](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/24378)|
|26618|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/26618](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/26618)|
|29337|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/29337](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/29337)|
|31719|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/31719](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/Airflow/pullrequest/31719)|

### Related repositories and infrastructure

|   |   |
|---|---|
|Resource|Link|
|Dora DataMart|[https://azdo.microlise.com/MicroliseCollection/Microlise/_git/DataMart.Academy.DoraMetrics](https://azdo.microlise.com/MicroliseCollection/Microlise/_git/DataMart.Academy.DoraMetrics)|
|Airflow deployment pipeline|[https://azdo.microlise.com/MicroliseCollection/Microlise/_release?_a=releases&view=mine&definitionId=970](https://azdo.microlise.com/MicroliseCollection/Microlise/_release?_a=releases&view=mine&definitionId=970)|
|Connection deployment pipeline|[https://azdo.microlise.com/MicroliseCollection/Microlise/_release?_a=releases&view=mine&definitionId=225](https://azdo.microlise.com/MicroliseCollection/Microlise/_release?_a=releases&view=mine&definitionId=225)|
|Slack alerts channel|[https://microlise.slack.com/archives/C09KD6S57CJ](https://microlise.slack.com/archives/C09KD6S57CJ)|

### Suggested training

**Course:** Async Programming in C#  
**Link:** [https://www.linkedin.com/learning/async-programming-in-c-sharp?u=94030082](https://www.linkedin.com/learning/async-programming-in-c-sharp?u=94030082)  
**Duration:** 3h 35m

**Conceptual mapping:** The C# course covers when to use `async`/`await` versus blocking calls, how to compose concurrent operations with `Task.WhenAll`, and how to avoid common pitfalls with shared state. These concepts map directly to the dataflow decisions in this pipeline — the mechanism differs (Airflow orchestration + synchronous Python vs C# async tasks), but the discernment skill is the same: identify independent work that can proceed concurrently, respect dependencies that require serialisation, and choose the right transport mechanism for data between steps.

---

_This document is standalone evidence for the Async Programming competency. It can be submitted directly to a competency portal or pasted into Confluence._