Extensión

Todo lo que viene integrado se registra a través de las mismas interfaces que puedes usar. Cuando el lenguaje no cubre un caso, extiéndelo en lugar de buscarle la vuelta.

Una transformación

import pyspark.sql.functions as F
from sparquet.transform.base import BaseTransformation

class NormalizeText(BaseTransformation):
    """Trims and lowercases a column."""

    def apply(self, df):
        column = self.config.params["column"]
        return df.withColumn(column, F.trim(F.lower(F.col(column))))

fw.register_transformation("normalize_text", NormalizeText)

{ "type": "normalize_text", "column": "email" }

self.config.params contiene todas las claves JSON excepto type y skip_if_false — lo que significa que skip_if_false funciona en tu transformación sin esfuerzo adicional.

Un validador

import pyspark.sql.functions as F
from sparquet.validation.base import BaseValidator, ValidationResult

class NoFutureDateValidator(BaseValidator):
    def validate(self, df):
        column = self.rule.params["column"]
        failed = df.filter(F.col(column) > F.current_date()).count()
        if failed:
            return ValidationResult("no_future_date", False, f"{failed} future dates", failed)
        return ValidationResult("no_future_date", True)

fw.register_validator("no_future_date", NoFutureDateValidator)

{ "type": "no_future_date", "column": "ordered_at" }

Devolver un ValidationResult — en lugar de lanzar una excepción — es lo que permite que on_failure decida la política y que el reporte registre el resultado.

Un conector

from sparquet.io.base import BaseReader, BaseWriter

class ElasticReader(BaseReader):
    def read(self):
        options = {**self.config.options, "es.resource": self.config.path}
        return self.spark.read.format("org.elasticsearch.spark.sql").options(**options).load()

class ElasticWriter(BaseWriter):
    def write(self, df):
        writer = df.write.format("org.elasticsearch.spark.sql").mode(self.config.mode)
        writer.options(**self.config.options).save(self.config.path)

fw.register_reader("elasticsearch", ElasticReader)
fw.register_writer("elasticsearch", ElasticWriter)

{ "format": "elasticsearch", "path": "orders/_doc", "options": { "es.nodes": "es.internal" } }

Nota

Los registros de lectores y escritores son a nivel de clase, compartidos por cada Sparquet del proceso. Los registros de transformaciones y validadores son por instancia, así que dos frameworks pueden llevar transformaciones personalizadas distintas.

Mantener Studio al día

Studio construye su paleta, sus formularios, su linter y el prompt del asistente a partir de un único catálogo en sparquet-studio/src/catalog/. Un tipo personalizado sigue funcionando sin tocarlo — los nodos desconocidos se importan, se preservan y se exportan sin cambios — pero no obtiene un formulario dedicado y el asistente no lo sugerirá.

Para convertirlo en ciudadano de primera clase, agrega una entrada:

src/catalog/transformations.core.ts

{
  type: 'normalize_text',
  label: 'Normalize text',
  family: 'compute',
  accent: 'transform',
  icon: 'Type',
  summary: 'Trims and lowercases a column.',
  description: 'Applies trim + lower to one column, in place.',
  fields: [
    {
      key: 'column',
      label: 'Column',
      type: 'text',
      required: true,
      placeholder: 'email',
      help: 'Column rewritten in place.',
    },
  ],
  keywords: ['trim', 'lower', 'clean'],
  gotchas: ['Rewrites the column in place — cast it first if the type matters.'],
  examples: [{ title: 'Normalize an email column', json: '{ "type": "normalize_text", "column": "email" }' }],
}

Esa única entrada te da el ítem de la paleta, el formulario del inspector, la regla de lint de campo obligatorio y una línea en el prompt de sistema del asistente.

Descubrir lo que un runtime realmente tiene

Los registros son dinámicos, así que la lista autoritativa vive en el proceso en ejecución. El runner local de Studio la expone:

curl -s localhost:8787/capabilities | jq

{
  "transformations": ["filter", "select", "…", "normalize_text"],
  "readers": ["parquet", "delta", "…", "elasticsearch"],
  "writers": ["parquet", "delta", "…", "elasticsearch"],
  "validators": ["not_null", "unique", "…", "no_future_date"]
}

Comparar eso con el catálogo es la forma más rápida de detectar un tipo personalizado que el editor todavía no conoce.

Contribuirlo de vuelta

Si la extensión es general — un conector para una base de datos común, una transformación que todo pipeline termina reescribiendo — envía un pull request. Dos cosas la hacen fusionable: un pipeline de ejemplo bajo examples/ que la ejercite, y una entrada de catálogo para que Studio también la incluya.