You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "isaiahiruoha (via GitHub)" <gi...@apache.org> on 2025/06/23 23:21:05 UTC
[PR] Add Airflow 3.0+ Task SDK support to AWS Batch Executor [airflow]
isaiahiruoha opened a new pull request, #52121:
URL: https://github.com/apache/airflow/pull/52121
---
Added Task SDK support for AWS BatchExecutor to enable compatibility with Airflow 3.0+.
The AWS BatchExecutor lacked support for handling Task SDK workloads, which has already been supported in the AWS ECSExecutor, changes were made to add this functionality. This meant the executor couldn't properly function with the latest Airflow architecture.
**Changes:**
- Implemented handling for `ExecuteTask` workloads in queue_workload method
- Added _process_workloads method to properly process Task SDK workloads
- Modified execute_async to handle cases where a workload object is passed instead of a direct command
- Added serialization logic to convert workloads to JSON for execution in AWS Batch containers
- Added new test case to verify Task SDK integration works correctly
**Testing:**
- Added unit test to verify proper handling of Task SDK workloads
- Successfully locally ran system tests with the updated executor
---
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Airflow 3.0+ Task SDK support to AWS Batch Executor [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #52121:
URL: https://github.com/apache/airflow/pull/52121#issuecomment-3002202206
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Airflow 3.0+ Task SDK support to AWS Batch Executor [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #52121:
URL: https://github.com/apache/airflow/pull/52121
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Airflow 3.0+ Task SDK support to AWS Batch Executor [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #52121:
URL: https://github.com/apache/airflow/pull/52121#discussion_r2162929126
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -97,6 +100,11 @@ class AwsBatchExecutor(BaseExecutor):
# AWS only allows a maximum number of JOBs in the describe_jobs function
DESCRIBE_JOBS_BATCH_SIZE = 99
+ if TYPE_CHECKING and AIRFLOW_V_3_0_PLUS:
+ # In the v3 path, we store workloads, not commands as strings.
+ # TODO: TaskSDK: move this type change into BaseExecutor
+ queued_tasks: dict[TaskInstanceKey, workloads.All] # type: ignore[assignment]
Review Comment:
I think I’ll take a shot after this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Airflow 3.0+ Task SDK support to AWS Batch Executor [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #52121:
URL: https://github.com/apache/airflow/pull/52121#issuecomment-2998753934
Need to fix CI, otherwise lgtm.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Airflow 3.0+ Task SDK support to AWS Batch Executor [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #52121:
URL: https://github.com/apache/airflow/pull/52121#discussion_r2162928677
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -97,6 +100,11 @@ class AwsBatchExecutor(BaseExecutor):
# AWS only allows a maximum number of JOBs in the describe_jobs function
DESCRIBE_JOBS_BATCH_SIZE = 99
+ if TYPE_CHECKING and AIRFLOW_V_3_0_PLUS:
+ # In the v3 path, we store workloads, not commands as strings.
+ # TODO: TaskSDK: move this type change into BaseExecutor
+ queued_tasks: dict[TaskInstanceKey, workloads.All] # type: ignore[assignment]
Review Comment:
Is this the time to do this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Airflow 3.0+ Task SDK support to AWS Batch Executor [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #52121:
URL: https://github.com/apache/airflow/pull/52121#discussion_r2164699638
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -97,6 +100,11 @@ class AwsBatchExecutor(BaseExecutor):
# AWS only allows a maximum number of JOBs in the describe_jobs function
DESCRIBE_JOBS_BATCH_SIZE = 99
+ if TYPE_CHECKING and AIRFLOW_V_3_0_PLUS:
+ # In the v3 path, we store workloads, not commands as strings.
+ # TODO: TaskSDK: move this type change into BaseExecutor
+ queued_tasks: dict[TaskInstanceKey, workloads.All] # type: ignore[assignment]
Review Comment:
There are many of the boiler plate methods that can be moved eventually too like queuing and processing the workloads. So far most executors just are copying a default implementation that could live on the base class and those executors that want to do anything fancy can still override them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org