[dag] dag by dag set AWS_MAX_ATTEMPTS?

[seunggabi]

#23475

I want to set each dag AWS_MAX_ATTEMPTS.
can I use this function?

[Taragolis]

Yes you can achieve this by different ways: see documentation: https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/connections/aws.html#avoid-throttling-exceptions

[Taragolis]

amazon provider in most cases just a wrapper around boto3/botocore client. So you can use their environment variables.

[seunggabi]

related to #23475

DefaultEmrStepSensor (this is my solution)

import boto3
from botocore.config import Config

import random
from airflow.providers.amazon.aws.sensors.emr_step import EmrStepSensor

_POKE_INTERVAL = 60
_RANDOM = {"START": 1, "END": 60, "INTERVAL": 1}
_MAX_ATTEMPTS = 15

class DefaultEmrStepSensor(EmrStepSensor):
    def __init__(
            self,
            poke_interval=None,
            **kwargs,
    ):
        poke_interval = poke_interval or _POKE_INTERVAL + random.randrange(
            _RANDOM["START"], _RANDOM["END"], _RANDOM["INTERVAL"]
        )
        super().__init__(poke_interval=poke_interval, **kwargs)

    def get_emr_response(self):
        config = Config(retries={"max_attempts": _MAX_ATTEMPTS, "mode": "standard"})
        emr_client = boto3.client("emr", config=config)

        self.log.info("Poking step %s on cluster %s", self.step_id, self.job_flow_id)

        return emr_client.describe_step(
            ClusterId=self.job_flow_id,
            StepId=self.step_id,
        )