Previsão do Tempo

Descrição do Projeto

Pipeline automatizado que coleta previsões meteorológicas de 7 dias para localidades específicas via API pública, armazena os dados em um Data Lake local, e os organiza para análises futuras.

O processo é completamente orquestrado pelo Apache Airflow, executando automaticamente toda segunda-feira às 8h, garantindo consistência e rastreabilidade dos dados climáticos ao longo do tempo.

Objetivos

Arquitetura do Pipeline

┌──────────────────┐
│   API Pública    │
│  (Meteorologia)  │
│  OpenWeather/    │
│  WeatherAPI      │
└────────┬─────────┘
         │ HTTP GET
         ▼
┌──────────────────┐
│ Apache Airflow   │
│  · Scheduler     │
│  · Extração      │
│  · Validação     │
│  · Transformação │
└────────┬─────────┘
         │ Save JSON
         ▼
┌──────────────────┐
│   Data Lake      │
│    (Local)       │
│ /data/weather/   │
│  └── YYYY-MM-DD/ │
│      └── data.json
└────────┬─────────┘
         │ Read
         ▼
┌──────────────────┐
│  Pandas/Python   │
│   Análises       │
│  (Futuro: BI)    │
└──────────────────┘

Fluxo Passo a Passo

1. Agendamento: DAG executa toda segunda-feira às 8h
2. Extração: Requisição GET para API de previsão do tempo
3. Validação: Verifica status 200 e estrutura JSON válida
4. Transformação: Extrai campos relevantes e formata datas
5. Armazenamento: Salva JSON particionado por data no Data Lake
6. Logging: Registra status da execução para monitoramento

Tecnologias Utilizadas

Exemplo de Código

DAG do Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import requests, json, os

default_args = {
    'owner': 'data_team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

def extract_weather_data(**context):
    API_KEY = os.getenv('WEATHER_API_KEY')
    CITY = 'Sao Paulo'
    url = f'https://api.openweathermap.org/data/2.5/forecast'
    params = {'q': CITY, 'appid': API_KEY, 'units': 'metric'}
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()

def save_to_datalake(**context):
    data = context['ti'].xcom_pull(task_ids='extract_weather')
    today = datetime.now().strftime('%Y-%m-%d')
    path = f'/opt/data_lake/weather/{today}'
    os.makedirs(path, exist_ok=True)
    with open(f'{path}/forecast.json', 'w') as f:
        json.dump(data, f, indent=2)

with DAG(
    'weather_pipeline',
    default_args=default_args,
    schedule_interval='0 8 * * 1',
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    extract = PythonOperator(task_id='extract_weather', python_callable=extract_weather_data)
    save = PythonOperator(task_id='save_to_datalake', python_callable=save_to_datalake)
    extract >> save

Exemplo de Resposta da API

{
  "city": { "name": "São Paulo", "country": "BR" },
  "list": [{
    "dt": 1704096000,
    "main": {
      "temp": 25.3,
      "feels_like": 26.1,
      "humidity": 72
    },
    "weather": [{ "main": "Clouds", "description": "nuvens dispersas" }],
    "wind": { "speed": 3.5 },
    "dt_txt": "2024-01-01 12:00:00"
  }]
}

Desafios e Soluções

Resultados

Possíveis Expansões

• Adicionar múltiplas cidades para comparação regional
• Integrar com PostgreSQL para queries SQL estruturadas
• Criar dashboard de visualização com histórico
• Implementar alertas para condições climáticas extremas
• Análise de acurácia: comparar previsões com dados reais

Aprendizados

• Como consumir APIs REST de forma confiável com tratamento de erros
• Importância do particionamento de dados para organização e performance
• Uso de Airflow XComs para passar dados entre tasks
• Boas práticas de gerenciamento de credenciais em pipelines