1
resposta

Erro ao executar o DAG, tarefa_2

Estou com erro ao executar o DAG Log dag Meu código:

from os.path import join

import pandas as pd
import pendulum
from airflow import DAG
from airflow.macros import ds_add
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator

with DAG(
    "dados_climaticos",
    start_date=pendulum.datetime(2023, 1, 22, tz="UTC"),
    schedule_interval='0 0 * * 1', # executar toda segunda feira
) as dag:

    tarefa_1 = BashOperator(
        task_id = 'cria_pasta',
        bash_command = 'mkdir -p  "/Users/francilenesilva/Documents/airflow-alura/semana={{data_interval_end.strftime("%Y-%m-%d")}}" '

    )

    def extrai_dados(data_interval_end):
        city = 'Boston'
        key = 'ANZQ5K8QQP8BXZ85F4ZEQ2FPK'

        URL = join('https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/{city}/{data_interval_end}/{ds_add(data_interval_end, 7)}?unitGroup=metric&include=days&key={key}&contentType=csv')    
        dados = pd.read_csv(URL)

        file_path = f'/Users/francilenesilva/Documents/airflow-alura/semana={{data_interval_end}/

        dados.to_csv(file_path + 'dados_brutos.csv')
        dados[['datetime','tempmin', 'temp', 'tempmax']].to_csv(file_path + 'temperaturas.csv')
        dados[['datetime', 'description', 'icon']].to_csv(file_path + 'condicoes.csv')

    tarefa_2 = PythonOperator(
        task_id = 'extrai_dados',
        python_callable = extrai_dados,
        op_kwargs = {'data_interval_end': '{{data_interval_end.strftime("%Y-%m-%d")}}'}
    )

    tarefa_1 >> tarefa_2

Estou usando o Mac M1 e optei por não usar máquina virtual.

1 resposta

Oii Francilene, tudo bem contigo?

Analisando seu código eu notei dois erros no conteúdo da variável file_path:

  • você colocou semana={{data_interval_end}, no entanto o correto é semana={data_interval_end} com apenas um par de chaves;
  • você esqueceu de fechar as aspas simples file_path = f'/Users/francilenesilva/Documents/airflow-alura/semana={data_interval_end}/'

Após corrigir esses erros, salve seu arquivo, exclua o histórico de execução do seu DAG pela interface do Airflow clicando naquela lixeirinha e espere ele aparecer novamente na interface para executá-lo e testar se vai funcionar corretamente :)

Qualquer dúvida estou à disposição. Bons estudos!