Update fix/clean up #4066
Update fix/clean up #4066
Conversation
📝 WalkthroughWalkthroughThe changes in this pull request introduce new documentation strings and enhance the functionality of Airflow DAGs related to daily measurements in BigQuery. Three new documentation strings are added to describe processes for cleaning and aggregating device measurements. Additionally, the DAGs are updated to include these documentation strings and improve error handling by modifying task decorators to allow retries and provide context. Changes
Possibly related PRs
Suggested reviewers
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Outside diff range and nitpick comments (3)
src/workflows/dags/dag_docs.py (1)
78-90: Documentation could be more comprehensiveThe documentation would benefit from additional details in the following areas:
- Notes section is empty but could include important information about:
- The criteria for identifying duplicates
- The retention policy for removed data
- The impact on downstream processes
- Purpose section could elaborate on why duplicate removal is necessary
src/workflows/dags/daily_measurements.py (2)
Line range hint
78-81: Add retry logic to resample taskThe resample task is missing retry logic while extract and load tasks have it. Since it performs data transformations that could fail, it should have similar error handling capabilities.
Apply this diff to add retry logic:
- @task() + @task(retries=3, retry_delay=timedelta(minutes=5)) def resample(data: pd.DataFrame):
Line range hint
119-122: Standardize task configurations across the DAGThere are inconsistencies in the task configurations:
- The resample task is missing retry logic
- The load task is missing provide_context=True compared to other DAGs
Apply these diffs to standardize the configurations:
- @task() + @task(retries=3, retry_delay=timedelta(minutes=5)) def resample(data: pd.DataFrame): - @task(retries=3, retry_delay=timedelta(minutes=5)) + @task(provide_context=True, retries=3, retry_delay=timedelta(minutes=5)) def load(data: pd.DataFrame):Also applies to: 125-126
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
src/workflows/dags/dag_docs.py(1 hunks)src/workflows/dags/daily_measurements.py(5 hunks)
🔇 Additional comments (1)
src/workflows/dags/daily_measurements.py (1)
4-9: LGTM! Well-structured imports
The imports are properly organized and align with the new functionality being added.
| daily_devices_measurements_historical_doc = """ | ||
| ### AirQo daily measurements data clean up - historical | ||
| #### Purpose | ||
| Aggregate daily device measurements using hourly devices measurements stored in bigquery going back a couple of days. | ||
| #### Notes | ||
|
|
||
|
|
||
| Data sources: | ||
| - BigQuery: daily_device_measurements | ||
| Data Destinations: | ||
| - BigQuery: daily_device_measurements | ||
| - <a href="https://airqo.africa/" target="_blank">AirQo</a> | ||
| """ |
There was a problem hiding this comment.
Fix title and add historical processing details
Similar to the realtime documentation, this needs attention:
- The title incorrectly mentions "data clean up"
- The Notes section should include:
- The historical data retention period
- How backfilling works
- Any special handling for historical data
Apply this diff to fix the title:
-### AirQo daily measurements data clean up - historical
+### AirQo historical daily measurements aggregationCommittable suggestion skipped: line range outside the PR's diff.
| daily_devices_measurements_realtime_doc = """ | ||
| ### AirQo daily measurements data clean up | ||
| #### Purpose | ||
| Aggregate daily device measurements using hourly devices measurements stored in bigquery | ||
| #### Notes | ||
|
|
||
|
|
||
| Data sources: | ||
| - BigQuery: daily_device_measurements | ||
| Data Destinations: | ||
| - BigQuery: daily_device_measurements | ||
| - <a href="https://airqo.africa/" target="_blank">AirQo</a> | ||
| """ |
There was a problem hiding this comment.
Fix incorrect title and enhance documentation
There are a few issues with this documentation:
- The title incorrectly states "data clean up" when the purpose is about aggregation
- The Notes section could include:
- The aggregation method used
- The frequency of updates
- Any data quality checks performed
Apply this diff to fix the title:
-### AirQo daily measurements data clean up
+### AirQo daily measurements aggregationCommittable suggestion skipped: line range outside the PR's diff.
| @task( | ||
| provide_context=True, | ||
| retries=3, | ||
| retry_delay=timedelta(minutes=5), | ||
| ) |
There was a problem hiding this comment.
Add retry logic to cleanup_and_load task
While the extract task has proper retry logic, the cleanup_and_load task is missing similar error handling. Since it performs database operations, it should also have retry capabilities.
Apply this diff to add retry logic:
- @task()
+ @task(retries=3, retry_delay=timedelta(minutes=5))
def cleanup_and_load(data: pd.DataFrame):Also applies to: 41-42
Description
[Provide a brief description of the changes made in this PR]
Enable re-streaming of daily device measurements using the original execution date.
Summary by CodeRabbit