Airflow with ECS task operator (Fargate Spot)

run_task = EcsRunTaskOperator( task_id="run_task", cluster=existing_cluster_name, task_definition=register_task.output, overrides={ "containerOverrides": [ { "name": container_name, "command": ["echo hello world"], }, ], }, network_configuration={"awsvpcConfiguration": {"subnets": existing_cluster_subnets}}, awslogs_group=log_group_name, awslogs_region=aws_region, awslogs_stream_prefix=f"ecs/{container_name}", )

hello_world = EcsRunTaskOperator( task_id="hello_world", cluster=cluster_name, task_definition=task_definition_name, launch_type="FARGATE", # Fargate로 변경 overrides={ "containerOverrides": [ { "name": container_name, "command": ["echo", "hello", "world"], }, ], }, network_configuration={ "awsvpcConfiguration": { "subnets": test_context[SUBNETS_KEY], "securityGroups": test_context[SECURITY_GROUPS_KEY], "assignPublicIp": "ENABLED", }, }, )

def __init__( self, *, task_definition: str, cluster: str, overrides: dict, launch_type: str = "EC2", capacity_provider_strategy: list | None = None, # 여기 group: str | None = None, placement_constraints: list | None = None, placement_strategy: list | None = None, platform_version: str | None = None, network_configuration: dict | None = None, tags: dict | None = None, awslogs_group: str | None = None, awslogs_region: str | None = None, awslogs_stream_prefix: str | None = None, awslogs_fetch_interval: timedelta = timedelta(seconds=30), propagate_tags: str | None = None, quota_retry: dict | None = None, reattach: bool = False, number_logs_exception: int = 10, wait_for_completion: bool = True, waiter_delay: int = 6, waiter_max_attempts: int = 1000000, # Set the default waiter duration to 70 days (attempts*delay) # Airflow execution_timeout handles task timeout deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), **kwargs, ): super().__init__(**kwargs)

def _start_task(self): ... run_opts["capacityProviderStrategy"] = self.capacity_provider_strategy

capacityProviderStrategy=[ { 'capacityProvider': 'string', 'weight': 123, 'base': 123 }, ],

prs_etl_pipeline_task = EcsRunTaskOperator( task_id='etl_pipeline_task', dag=etl_dag, on_failure_callback=task_fail_alert, aws_conn_id=aws_conn_id, cluster='etl-pipeline', task_definition='etl_pipeline', launch_type='FARGATE', overrides={}, wait_for_completion=True, capacity_provider_strategy=[{ 'capacityProvider': 'FARGATE_SPOT', 'weight': 1, 'base': 1 }], # datapipeline vpc network_configuration={ "awsvpcConfiguration": { "subnets": ["subnet-xxxcaxxxxxxxxx"], "securityGroups": ["sg-093xxxxxxx"], "assignPublicIp": "ENABLED" } }, awslogs_group=log_group_name, awslogs_region=aws_region, awslogs_stream_prefix=awslogs_stream_prefix, )