Repaso proceso dbt

Fase 1. Explorar los datos y redactar el informe de calidad

Echar un vistazo a los datos, para ver si hay datos erróneos, que faltan, etc. En mi opinión, lo primero que hay que hacer siempre independientemente del tipo de proyecto que tengamos.

Fase 2. Fuentes, seed, macros y staging

Creamos un entorno virtual

python -m venv .venv

Lo activamos

.venv\Scripts\activate

Instalamos dbt-duckdb

pip install dbt-duckdb

Creamos el proyecto:

dbt init bicing

Vamos a la carpeta:

cd bicing

Inicializamos git:

git init

git add .

git commit -m Inicio

Arrancamos el Visual Studio Code (opcional)

code .

Copiamos los datos csv a la carpeta datos, distritos.csv a seed y el carrega_dades.sql. El carregar.py a scripts\

python .\scripts\carregar.py

Cargamos los distritos:

dbt seed

Eliminamos models\example y también de dbt_project

Usamos dbt debug para comprobar si todo está bien

Stagging: Tenemos tres fuentes, pues creamos las carpetas y un sources.yaml

models
– staging
– intermeadiate
– marts

sources.yml

version: 2

sources:
  - name: raw
    description: "Datos originales del ejercicio, cargados desde CSV en el esquema raw. Todas las columnas llegan como texto."
    schema: raw
    tables:
      - name: estaciones
        description: "Una fila por estación y carga. Puede haber estaciones repetidas y filas de prueba."
      - name: uso
        description: "Viajes de salida y de llegada por estación y mes. Puede haber duplicados y claves rotas."

stg_distritos

select * 
from {{ ref('distritos') }}

stg_estaciones

select station_id as id_estacion,
name  as nombre_estacion,
type  as tipo_estacion,
lat as latitud,
lon as longitud,
altitude as altitud,
address as direccion,
post_code  as codigo_postal,
capacity as capacidad,
district  as distrito,
status as estado,
installed_on  as fecha_instalacion,
loaded_at  as fecha_carga
from {{ source('raw', 'estaciones') 

stg_uso

select station_id as id_estacion,
period as periodo,
departures as salidas,
arrivals as llegadas,
loaded_at as fecha_carga
from {{ source('raw', 'uso') }}

Usaremos una macro para normalizar texto
macros
– normalizar_texto.sql

{% macro normalizar_texto(columna) %}
    nullif(
        strip_accents(lower(trim({{ columna }}))),
        ''
    )
{% endmacro %}

La usaremos en stg_estaciones:

select station_id as id_estacion,
{{ normalizar_texto('name')}}  as nombre_estacion,
{{ normalizar_texto('type')}}  as tipo_estacion,
lat as latitud,
lon as longitud,
altitude as altitud,
address as direccion,
post_code  as codigo_postal,
capacity as capacidad,
district  as distrito,
status as estado,
installed_on  as fecha_instalacion,
loaded_at  as fecha_carga
from {{ source('raw', 'estaciones') }}

Fase 3. Intermediate: reglas de calidad y cruces

Sirve para realizar cálculos que puedan ser útiles en siguientes fases. Relacionan tablas, hacen sumas, enriquecen los datos. Puedo tener una tabla ‘pedidos’ con el id cliente sin el nombre, o puedo calcular el total por clientes.

En este caso vamos a calcular el total de salidas y llegadas por estación y mes:

intermediate
– int_total_por_mes

select id_estacion,
extract(month from periodo) as mes,
sum(salidas) as total_salidas,
sum(llegadas) as total_llegadas
from stg_uso
group by id_estacion,
extract(month from periodo) 

Aquí nos encontramos varios problemas que nos obligarán a limpiar los datos. Por un lado convertiremos el periodo en fecha con una macro:

normalizar_periodo.sql

{% macro normalizar_periodo(columna) %}
    coalesce(
        try_strptime(trim(cast({{ columna }} as varchar)), '%Y-%m'),
        try_strptime(trim(cast({{ columna }} as varchar)), '%m/%Y')
    )::date
{% endmacro %}

Y por otro cambiaremos el stg_uso:

select station_id as id_estacion,
{{normalizar_periodo('period')}} as periodo,
try_cast(departures as integer) as salidas,
try_cast(arrivals as integer) as llegadas,
loaded_at as fecha_carga
from {{ source('raw', 'uso') }}

 

Fase 4. Marts: indicadores finales

Hacemos un informe con la cantidad de salidas y llegadas de cada estación:
fct_totales_estacion

with estaciones as (
    select
        *
    from {{ ref('stg_estaciones') }}
    where id_estacion is not null
),
uso as (select * from
{{ref('int_total_por_mes')}}
)
select 
e.nombre_estacion,
sum(total_salidas) as total_salidas,
sum(total_llegadas) as total_llegadas
 from
estaciones e join
uso u on
e.id_estacion=u.id_estacion
group by e.nombre_estacion

Fase 5. Jinja: variables y bucles

Fase 6. Tests y documentación