气流 DAG 失败 - 错误 - HTTP 错误 404：未找到

Question

我是 Apache Airflow 的菜鸟。从字面上看，刚刚开始，我遇到了一个错误。我写了我的第一个 dag，并且正在调用 Python 脚本。当我设置它时，它最初成功运行并工作，我安排它每天运行一次。我今天来检查它，dag 失败并显示消息 ERROR - HTTP Error 404: Not Found。

一切对我来说都是新的，如果这是一个简单的解决方法，我很抱歉，但我不明白为什么我会收到 404 错误。我试过重启 docker 看看是否是网络服务器问题，但没有运气。

感谢您的帮助

Screenshot

DAG

from airflow.models import DAG
from datetime import datetime, timedelta
from airflow.operators.python_operator import PythonOperator
from covid_cases import covid_data


default_args = {
    'owner': 'airflow',
    'start_date': datetime(2020, 10, 4),
    'retries': 2,
    'retry_delay': timedelta(seconds=20)}

dag =  DAG(dag_id = 'covid_updates',
           default_args = default_args,
           schedule_interval = "0 4 * * *")

t1 = PythonOperator(task_id = 'covid_update',
                    python_callable = covid_data,
                    dag = dag)

t1

PythonOperator - covid_cases.py

def covid_data():
    """

    Overview:
    ---------
    Downloads the USA COVID data directly from John Hopkins CSSEGISandData.
    This function merges all data from most current date to earliest date (2020-4-11).
    Using this function a user can conduct time series analysis in how COVID
    increases/decreases in various states.

    Output:
    -------
    One uncleaned .csv file called "usa_covid_cases.csv"


    """
    from datetime import datetime, timedelta
    import pandas as pd
    from urllib.error import HTTPError

    # Set starting index
    i = 1

    # Earliest dataset available on GitHub
    start_date = datetime.strptime('2020-4-11', '%Y-%m-%d').date()

    # Pulling today's date minus 1 day due to delay posting on GitHub
    today = datetime.now().date() - timedelta(days=i)

    # Setting llist to store dataframe file names
    file_names = []

    # Looping until date is equal to earlist date = Start Date
    while not (start_date.day == today.day and start_date.month == today.month and start_date.year == today.year):

        # Extracting variables from current date
        day = today.day
        month = today.month
        year = today.year

        # Cleaning and converting values for formatting on GitHub URL link
        if day < 10:
            day = '0' + str(day)

        if month < 10:
            month = '0' + str(month)

        # Setting variable for each url
        url = 'https://raw.githubusercontent.com/CSSEGISandData/COVID-19/master/csse_covid_19_data/csse_covid_19_daily_reports_us/{}-{}-{}.csv'\
                    .format(month, day, str(year))

        try:
            # Reading each url as a datafra,
            df = pd.read_csv(url, error_bad_lines=False)
        except HTTPError as e:
            # handle the error (print, log, etc)
            continue
        finally:
            # Code moved here to prevent an endless loop
            # Subtracting the new index to increase 1 less day from the current date
            today = datetime.now().date() - timedelta(days=i)

        # Saving each dataframe into the empty list
        file_names.append(df)

        # Increasing the index by 1
        i += 1

    # Once while loop ends - concat all the files into a single dataframe
    new_df = pd.concat(file_names)

    # Save output into new csv file
    new_df.to_csv('usa_covid_cases.csv')

Answer 1

从日志中可以看出，当您尝试从 URL 读取数据并且该 URL 不存在时会发生错误。

pd.read_csv(url, error_bad_lines=False) #Line 50 of covid_data.py

Answer 2

我刚刚看了GitHub repo，最早的数据集来自04-12-2020。

为防止 DAG 从可能已删除的数据集中失败，您可以将 pd.read_csv() 包装在 try except 块中，如下所示：

# Import the HTTPError (error from you screenshot)
from urllib.error import HTTPError
...

try:
    # Reading each url as a datafra,
    df = pd.read_csv(url, error_bad_lines=False)
except HTTPError as e:
    # handle the error (print, log, etc)
    continue
finally:
    # Code moved here to prevent an endless loop
    # Subtracting the new index to increase 1 less day from the current date
    today = datetime.now().date() - timedelta(days=i)

 # Saving each dataframe into the empty list
 file_names.append(df)

 # Increasing the index by 1
 i += 1

...