Projeto: ETL com Apache Airflow usando a API de Dados Abertos da Assembleia Legislativa de Minas

2. Requisitos Funcionais

Coleta de dados

O sistema vai coletar dados do endpoint:

• https://dadosabertos.almg.gov.br/ws/proposicoes/pesquisa/direcionada?tp=100&formato=json&ano=2024&ord=3&p=50000&ini=20250123&fim=20250126, utilizando os filtros de paginação, data de início e data de fim.
• https://dadosabertos.almg.gov.br/ws/proposicoes/pesquisa/direcionada?formato=json&ano=2024&num=11063: URL de exemplo para fazer o reprocessamento.

Processamento de dados

O sistema deverá realizar o tratamento de caracteres, como:

• Remover espaços desnecessários;
• Remover caracteres de quebra de linha, deixando o texto em uma única linha;
• Formatar datas corretamente.

Armazenamento de dados

Os dados serão armazenados no banco SQL Server.

3. Requisitos Não Funcionais

• Execução: O sistema deve rodar com um intervalo de um dia para envio de dados.
• Monitoramento: O sistema deve incluir logs de execução, registrando históricos e erros para reprocessamento.

4. Estrutura do Banco de Dados

Tabela proposicao

CREATE TABLE proposicao (
    ID INTEGER IDENTITY(1,1) PRIMARY KEY,
    AUTOR NVARCHAR(300),
    DATA_PRESENTACAO DATETIME,
    EMENTA NVARCHAR(MAX),
    REGIME VARCHAR(50),
    SITUACCAO VARCHAR(80),
    TIPO_PROPOSICAO VARCHAR(100),
    NUMERO VARCHAR(40) UNIQUE,
    ANO INTEGER,
    CIDADE VARCHAR(60),
    ESTADO VARCHAR(80),
    DATA_INSERSAO_REGISTRO DATETIME DEFAULT GETDATE(),
    DATA_ATUALIZACAO_REGISTRO DATETIME
);

Dicionário de dados da tabela proposicao

Tabela tramitacao

CREATE TABLE tramitacao (
    ID INTEGER IDENTITY(1,1) PRIMARY KEY,
    DESCRICAO VARCHAR(MAX),
    LOCAL_PROPOSICAO VARCHAR(60),
    ID_PROPOSICAO VARCHAR(40),
    DATA_CRIACAO_TRAMITACAO DATETIME,
    DATA_CRIACAO_REGISTRO DATETIME DEFAULT GETDATE(),
    DATA_ATUALIZACAO_REGISTRO DATETIME,
    FOREIGN KEY (ID_PROPOSICAO) REFERENCES proposicao(NUMERO)
);

Dicionário de dados da tabela tramitação

Tabela log_dag

CREATE TABLE log_dag (
    ID INTEGER IDENTITY(1,1) PRIMARY KEY,
    TIPO_ERROR INTEGER,
    TIPO_LOG VARCHAR(20),
    URL_API VARCHAR(MAX),
    MENSAGEM_LOG VARCHAR(300),
    JSON_XML VARCHAR(MAX),
    JSON_ENVIO VARCHAR(MAX),
    DATA_REGISTRO DATETIME DEFAULT GETDATE()
);

Dicionário de dados da tabela log_dag

Tabela dag_error

CREATE TABLE dag_error (
    ID INTEGER IDENTITY(1,1) PRIMARY KEY,
    TIPO_ERROR INTEGER,
    NUMERO VARCHAR(40),
    URL_API VARCHAR(MAX),
    JSON_XML VARCHAR(MAX),
    JSON_ENVIO VARCHAR(MAX),
    MENSAGEM_ERRO VARCHAR(300),
    DATA_REGISTRO DATETIME DEFAULT GETDATE(),
    DATA_ATUALIZACAO DATETIME
);

Dicionário de dados da tabela dag_error

5. Fluxo da DAG

A DAG inicia verificando a conexão com o banco de dados e a API. Se ambas estiverem ativas, inicia-se o processo ETL para extração e tratamento de dados. Caso algum registro falhe, ele é gravado na tabela dag_error e reprocessado antes da finalização do ETL.

6. Diagrama de Classe

A figura abaixo, mostra uma proposta de diagrama de classe, com destaque para a a conexão da API e banco de dados, podendo ser substituído sem prejuízo ao funcionamento do processo de ETL.

7 - Assistir ao vídeo

Link do reposítório