Dodawanie ponownych prób

Zauważasz, że pewien DAG często kończy się niepowodzeniem na zadaniu pobierającym dane z określonego źródła. Co irytujące, uruchomienie tego zadania kilka minut później zazwyczaj rozwiązuje problem. Po zapoznaniu się z funkcją ponownych prób w DAG-ach Airflow postanawiasz ją zastosować, żeby nie musieć ręcznie restartować zadania.

dag, task i timedelta są już zaimportowane.

To ćwiczenie jest częścią kursu

Wprowadzenie do Apache Airflow w Pythonie

Zobacz kurs

Instrukcje do ćwiczenia

Ustaw zadanie extract_data tak, aby wykonało 3 ponowne próby przed zgłoszeniem błędu.
Dodaj 10-minutowe opóźnienie między kolejnymi próbami dla zadania extract_data.

Interaktywne ćwiczenie praktyczne

Spróbuj tego ćwiczenia, uzupełniając ten przykładowy kod.

@dag(schedule="@daily", start_date=datetime(2026, 5, 1))
def etl_pipeline():

  # Set retries and retry delay on extract_data
  @task(____=3, ____=____(minutes=10))
  def extract_data():
    print("Extracting data from source...")
  
  @task()
  def process_source_data():
    print("Now processing data...")

  extract_data() >> process_source_data()
  
etl_pipeline()

Edytuj i uruchom kod