ETL com Python e geoinformação: pipelines de produção
Um pipeline de dados não é apenas um script que move arquivos. É uma cadeia de decisões técnicas que define se uma organização pode confiar em seus indicadores, mapas e produtos finais. Este artigo explica como projetar um pipeline que funcione em produção.
O que torna um pipeline ETL robusto?
A maioria dos pipelines que vi em projetos reais falha pelo mesmo motivo: não há regras explícitas de validação nem registro do que aconteceu em cada etapa. Quando algo falha, não é possível saber onde nem por quê.
Um pipeline robusto tem três propriedades: verificabilidade (posso saber se os dados estão corretos), rastreabilidade (posso saber de onde veio cada registro) e manutenibilidade (posso modificá-lo sem medo de quebrar algo).
A arquitetura em camadas
A decisão mais importante em um pipeline é nunca sobrescrever os dados de origem. A camada raw armazena uma cópia exata do que chegou, com timestamp de ingestão. Se algo falhar nas etapas seguintes, é possível reprocessar sem pedir o arquivo novamente à fonte.
A camada staging é onde ocorrem as transformações: normalização de campos, correção de codificações, conversão de sistemas de referência, joins e agregações. Os erros também são registrados aqui — não são descartados, são documentados.
A camada producción contém apenas registros que passaram por todas as validações. É a única consumida pelos sistemas de análise, mapas e APIs.
Validação explícita em geoinformação
Em dados geoespaciais, validar significa mais do que verificar se um campo não é nulo. Um polígono pode ser sintaticamente correto, mas topologicamente inválido (autointerseção). Uma coordenada pode estar dentro do intervalo numérico, mas fora da área de estudo. Um atributo pode existir, mas não pertencer ao domínio permitido.
Implementação com Python e PostGIS
O padrão a seguir mostra uma estrutura de pipeline real com validação explícita e logging estruturado:
import logging
from datetime import datetime
from sqlalchemy import create_engine, text
import geopandas as gpd
from shapely.validation import make_valid
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s"
)
log = logging.getLogger(__name__)
ENGINE = create_engine("postgresql://user:pass@localhost/geodata")
def ingest_raw(filepath: str, layer: str) -> gpd.GeoDataFrame:
"""Carga el archivo original sin modificar y registra la ingesta."""
gdf = gpd.read_file(filepath, layer=layer)
gdf["_ingested_at"] = datetime.utcnow()
gdf["_source_file"] = filepath
gdf.to_postgis("raw_parcels", ENGINE, if_exists="append", index=False)
log.info("raw: %d registros ingestados desde %s", len(gdf), filepath)
return gdf
def validate(gdf: gpd.GeoDataFrame) -> tuple[gpd.GeoDataFrame, list[dict]]:
"""Aplica reglas de validación y separa registros válidos de errores."""
errors = []
valid_idx = []
for idx, row in gdf.iterrows():
record_id = row.get("id", idx)
geom = row.geometry
# Regla 1: geometría no nula
if geom is None or geom.is_empty:
errors.append({"id": record_id, "field": "geometry", "reason": "null_or_empty"})
continue
# Regla 2: geometría válida (auto-intersecciones, etc.)
if not geom.is_valid:
fixed = make_valid(geom)
if not fixed.is_valid:
errors.append({"id": record_id, "field": "geometry", "reason": "invalid_geometry"})
continue
gdf.at[idx, "geometry"] = fixed # corregir si es posible
# Regla 3: CRS correcto
if gdf.crs is None or gdf.crs.to_epsg() != 4326:
errors.append({"id": record_id, "field": "crs", "reason": f"expected_4326_got_{gdf.crs}"})
continue
# Regla 4: atributos obligatorios
for field in ("parcel_id", "land_use", "area_m2"):
if row.get(field) is None:
errors.append({"id": record_id, "field": field, "reason": "required_null"})
break
else:
valid_idx.append(idx)
log.info("validate: %d válidos, %d errores", len(valid_idx), len(errors))
return gdf.loc[valid_idx].copy(), errors
def load_staging(gdf: gpd.GeoDataFrame, errors: list[dict]) -> None:
"""Carga staging con datos válidos y registra errores."""
gdf.to_postgis("staging_parcels", ENGINE, if_exists="replace", index=False)
if errors:
import pandas as pd
err_df = pd.DataFrame(errors)
err_df["logged_at"] = datetime.utcnow()
err_df.to_sql("etl_errors", ENGINE, if_exists="append", index=False)
log.warning("staging: %d errores registrados en etl_errors", len(errors))
# Ejecución
raw = ingest_raw("parcelas_2026.gpkg", layer="parcelas")
valid_gdf, errs = validate(raw)
load_staging(valid_gdf, errs)
Rastreabilidade: saber o que aconteceu com cada registro
O logging anterior registra os erros, mas a rastreabilidade vai além: cada registro em produção deve ser rastreável até sua fonte original. Para isso, os campos _ingested_at e _source_file são propagados de raw até produção.
Em projetos geoespaciais complexos, nos quais os dados passam por múltiplas transformações (mudanças de CRS, dissolves, interseções espaciais), também é útil guardar o _etl_step — o nome da função ou do processo que produziu o registro em seu estado atual.
Regras para um pipeline duradouro
- Nunca sobrescreva raw. É seu seguro de dados. Se algo falhar, você pode reprocessar.
- As regras de validação são código, não comentários. Escreva-as como funções com nomes claros e testes unitários.
- Os erros são registrados, não ignorados. Um pipeline silencioso que descarta dados incorretamente é mais perigoso do que um que falha explicitamente.
- Cada etapa tem entrada esperada e saída verificável. Defina o schema de saída antes de escrever a transformação.
- PostGIS é seu aliado. Use
ST_IsValid,ST_MakeValid,ST_Withine restrições CHECK no banco para validar no momento da carga.
Um pipeline assim não é mais lento nem mais complexo do que um sem validação. Mas, quando algo falha — e em algum momento sempre falha — você tem exatamente as informações necessárias para entender o que aconteceu e como corrigir.