API Python e CLI

Sparquet

O ponto de entrada. Ele detém a sessão Spark e compartilha os engines de transformação e validação entre todas as execuções.

from sparquet import Sparquet

fw = Sparquet(spark={"app_name": "daily_load", "master": "local[*]"})

fw.run("pipelines/orders.json")
fw.run("pipelines/customers.json")

fw.stop()

Criar uma instância por processo e executar muitos pipelines através dela é o formato pretendido: a sessão é construída uma vez, e temp views escritas por um pipeline são visíveis para o próximo.

run(config_path, input_df=None, columns=None, params=None)

Executa um pipeline descrito por um arquivo.

Argumento	Tipo	Propósito
config_path	str	caminho para o arquivo JSON
input_df	DataFrame	substitui a leitura do input — o pipeline começa a partir deste DataFrame
columns	dict	colunas literais injetadas antes das transformações
params	dict	substituições de {param}

result = fw.run(
    "orders.json",
    params={"region": "br", "since": "2026-01-01"},
    columns={"load_id": "2026-01-15T03:00:00Z"},
)

columns adiciona cada entrada como uma coluna literal (F.lit(value)) logo após a leitura — a maneira limpa de carimbar um id de batch ou um timestamp de execução sem editar o arquivo.

run_from_dict(config, input_df=None, columns=None, params=None)

O mesmo, a partir de um dicionário já em memória:

result = fw.run_from_dict({
    "name": "inline",
    "input": {"format": "delta", "path": "sales.orders"},
    "output": {"format": "parquet", "path": "/tmp/orders"},
})

Útil quando um pipeline é gerado — por um notebook, um teste ou o runner local do Studio.

Registro

fw.register_reader("my_format", MyReader)
fw.register_writer("my_format", MyWriter)
fw.register_transformation("normalize_text", NormalizeText)
fw.register_validator("no_future_date", NoFutureDateValidator)

Veja Estendendo.

Nota

Os registries de reader e writer são a nível de classe: registrar um formato afeta todo Sparquet no processo. Os registries de transformação e validator são por instância.

stop()

Para a sessão Spark subjacente. No Databricks, onde a sessão é da plataforma, deixe-a rodando.

Pipeline

Um pipeline, sem o wrapper do framework — útil em testes e quando você mesmo gerencia a sessão.

from sparquet import Pipeline

result = Pipeline.from_file("orders.json").run()
result = Pipeline.from_dict({...}).run()

Pipeline também aceita engines injetados, que é como você testa um pipeline com uma transformação customizada registrada apenas para aquele caso.

PipelineResult

Toda execução retorna este objeto. Ele nunca lança exceção — falhas voltam como dados.

@dataclass
class PipelineResult:
    pipeline_name: str
    success: bool
    rows_read: int = 0
    rows_written: int = 0
    validation_results: list[ValidationResult] = []
    output_metrics: list[OutputMetrics] = []
    error: str | None = None
    output_df: DataFrame | None = None
    skipped: bool = False

    def summary(self) -> str: ...

Campo	Significado
success	False quando a execução falhou; verifique error para o motivo
skipped	True quando stop_if_empty encerrou a execução — success permanece True
rows_read	linhas lidas da entrada (0 quando um DataFrame foi injetado)
rows_written	total de linhas escritas — a soma das contagens por destino
validation_results	uma entrada por regra: rule_type, passed, message, failed_count, severity, metric_value, check_name
output_metrics	um OutputMetrics por destino (veja abaixo)
output_df	o DataFrame transformado, disponível quando input_df foi injetado
error	a mensagem de falha

output_metrics

Uma entrada por destino que o pipeline escreveu, em ordem:

@dataclass
class OutputMetrics:
    format: str
    path: str
    mode: str
    rows_written: int

Cada rows_written é contado no DataFrame final daquele destino — após suas próprias transformations e projeção de colunas, logo antes da escrita — então é exato mesmo quando uma cadeia por destino explode ou agrega linhas. PipelineResult.rows_written é a soma dessas contagens.

result = fw.run("orders.json")
for m in result.output_metrics:
    print(f"{m.format:8} {m.mode:9} {m.rows_written:>10}  {m.path}")

result = fw.run("orders.json")

if result.skipped:
    log.info("nothing to process")
elif not result.success:
    raise RuntimeError(result.error)
else:
    log.info(result.summary())
    for check in result.validation_results:
        if not check.passed:
            alert(f"{check.rule_type}: {check.message} ({check.failed_count} rows)")

Encadeando pipelines em um job

fw = Sparquet(spark={"app_name": "registration"})

for conf in ["conf_a.json", "conf_b.json", "conf_c.json"]:
    result = fw.run(conf, params={"dt_ref": dt_ref})
    if not result.success:
        raise RuntimeError(f"{conf}: {result.error}")

# every pipeline above wrote the same temp view; this one publishes it
final = fw.run("conf_commit.json")
fw.stop()

Cada pipeline escreve uma temp view de staging, e o último valida e publica. Como eles compartilham uma sessão, as views sobrevivem entre as chamadas.

CLI

python -m sparquet.cli pipeline.json

Ou, quando instalado como pacote, através do console script:

sparquet pipeline.json

A CLI parseia o arquivo, o executa e imprime o resultado estruturado. É o ponto de entrada certo para um scheduler ou um container: sem script wrapper para manter.

Logging estruturado

Cada linha de log é um objeto JSON no stderr:

{"timestamp": "2026-01-15T03:00:12.918Z", "level": "INFO", "message": "Pipeline finished", "pipeline": "orders_curated", "rows_written": 41233}

Esse formato cai direto no Datadog, CloudWatch ou Splunk sem um parser. O runner local do Studio captura os mesmos registros e os mostra no painel de Run.