24 KiB
Executable File
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.
Performance and dataflow constraints
-
Many independent API domains — builds, deployments, and releases do not depend on repository metadata and can run concurrently.
-
Strict cross-domain dependencies — commits require pull request IDs; pull requests require repository IDs; repositories require team/system enrichment.
-
Large payloads — tens of thousands of records per domain; inter-task data transfer must not use Airflow's metadata database.
-
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
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:
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:
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:
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:
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:
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:
-
Airflow already parallelises at the DAG layer — adding in-process async would overlap with worker-level concurrency without simplifying the dependency graph.
-
Predictable API rate-limit behaviour — sequential calls per worker are easier to reason about than concurrent requests from multiple async coroutines.
-
Simpler error handling and testability — each callable is a plain Python function with 80%+ unit test coverage across all domains.
-
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
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
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 usingtask groupsto orchestrate the flow of data, we originally usedxcomto 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
parquetsfor 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_folderthrough the taskclear_create_directorywhich calls the scriptClearAndCreateCollectionDirectory.
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
Work item: #108242
Problem (from PR description):
The task
clear_create_directorywas 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
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 | Added a schedule for the dora dag at 1 am | Operationalised daily concurrent collection window |
| 26618 | Adding a conversion to GUID from null repo guid values | Data integrity fix in repository enrichment chain |
| 29337 | Updating the SQ portfolio links - Dora DAG | Maintained serial dependency chain for system/repo enrichment |
| 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:
-
AZDO API rate-limit analysis to avoid throttling
-
Airflow worker memory profiling under concurrent requests
-
Updated unit tests for concurrent error paths
-
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:
-
Where to parallelise — independent AZDO domains via Airflow TaskGroup branches
-
Where to serialise — dependent domains via parquet handoffs and explicit
>>dependencies -
Where to stay synchronous — blocking API calls within tasks, because orchestration-layer concurrency already addresses the primary bottleneck
-
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
Related repositories and infrastructure
| Resource | Link |
| Dora DataMart | 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 |
| Connection deployment pipeline | https://azdo.microlise.com/MicroliseCollection/Microlise/_release?_a=releases&view=mine&definitionId=225 |
| Slack alerts channel | 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
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.