magmalake/magmalake-apps

Una aplicación magmalake es un proyecto pixi que contiene tres tipos de programas, todos en Mojo: pipelines que orca programa y que escriben tablas Iceberg, el propio orca del proyecto (programador e interfaz de usuario), y apps: servidores flare de larga duración, como una consola que lee las tablas. En AWS, una pila de CloudFormation ejecuta orca, Postgres, las apps y Caddy en una única instancia de control, y cada ejecución de pipeline en sus propias instancias.

Para pipelines en profundidad (particiones, claveadas, sensores, tareas dista), carga el skill orca; escribe Mojo con el skill mojo-syntax de Modular. Lo propio de mojoshelf: npx skills add mojoshelf/mojoshelf --skill mojoshelf-consume --yes.

Crear el proyecto

pixi init lake -c https://conda.modular.com/max -c conda-forge
cd lake
# En [workspace] de pixi.toml, añade:   preview = ["pixi-build"]
pixi add "mojo-compiler==1.2.0"
pixi workspace platform add linux-aarch64   # la plataforma de la imagen AWS
pixi shelf add orca-mojo
pixi add postgresql                          # un Postgres local, para desarrollo

• pixi shelf add orca-mojo (orca 0.3.0) añade orca y todo lo que necesita a pixi.toml, incluyendo dista-mojo, iceberg-mojo, flare_mlake, objectstore-mojo y json_mlake, así que importan de una vez. Añade otros tins con pixi shelf add <tin>. Toma minutos e imprime el registro de compilación de cada tin; no se ha colgado.
• linux-aarch64 debe estar en las plataformas, o orca aws push se niega a construir; la imagen se instala con pixi install --locked.
• Leer el código fuente de un tin: el entorno solo contiene paquetes .mojoc compilados para la mayoría de los tins (flare y json son las excepciones), así que lee el código fuente en GitHub (https://github.com/magmalake/<repo> en la rev de pixi.toml) o en la caché git de pixi: ls -d ~/Library/Caches/rattler/cache/git-v0/checkouts/*/<primeros 7 de rev>* en macOS, ~/.cache/rattler/cache/git-v0/checkouts/ en Linux. Las rutas de archivo citadas en estos skills están en esos checkouts.

Diseño

orca aws push construye exactamente este diseño:

pixi.toml, pixi.lock
bin/scheduler.mojo     el orca del proyecto: las dos líneas de abajo
bin/<nombre>.mojo        una app, p.ej. bin/console.mojo
pipelines/<nombre>.mojo  un pipeline, terminando en orca_main[...]()
src/                   código compartido, en la ruta de importación (-I src)
Dockerfile.orca        opcional: reemplaza el Dockerfile por defecto de orca

bin/scheduler.mojo es exactamente:

from orca.command import orca_command

def main() raises:
    orca_command()

Nunca lo nombres orca.mojo: ocultaría el paquete orca. En la imagen se convierte en /opt/orca/bin/orca, otros bin/<nombre>.mojo se convierten en /opt/orca/bin/<nombre>, y los pipelines en /opt/orca/pipelines/<nombre>.

El lago: tablas en un almacén

orca encuentra las tablas a través de ORCA_WAREHOUSE, la raíz de un catálogo de sistema de archivos Iceberg: el conjunto de datos demo.days es la tabla en <warehouse>/demo/days. El almacén es un directorio local o una ubicación S3 (s3://bucket/warehouse). En AWS, la pila lo coloca en el bucket público:

Variable	Valor en AWS	Quién la lee
ORCA_WAREHOUSE	s3://mlake-public-<account>/warehouse	pipelines (a través de orca.tables), apps, cualquiera sin firmar
ORCA_PUBLIC_WAREHOUSE	la misma ubicación	código que quiere decir "el público"

El bucket propio de orca permanece privado y solo contiene registros y estado de ejecución. Cualquiera puede leer el lago, sin firmar: aws s3 ls --no-sign-request s3://mlake-public-<account>/warehouse/, o DuckDB y pyiceberg en sus modos anónimos. Así que cada tabla que un pipeline escribe es pública.

Coloca las definiciones de tablas compartidas en src/lake.mojo, para que los pipelines y las apps coincidan en nombres y esquemas. Los ejemplos a continuación importan tres cosas de ahí, copiadas en su totalidad en references/lake.md:

• DAYS = "demo.days", el nombre del conjunto de datos;
• days_schema(), su esquema Iceberg (day string, campo 1; events long, campo 2), desde Schema.parse del JSON de Iceberg;
• load(namespace, name), una tabla desde ORCA_WAREHOUSE (por defecto build/warehouse), a través de FilesystemCatalog.local(root) o, para s3://, FilesystemCatalog(root, FileIO()).

Escribir tablas desde un paso

Un paso declara los conjuntos de datos que escribe (writes=[...]) y lee (reads=[...]); orca.tables rechaza cualquier otro. Usa ensure_table para crear una tabla en el primer uso y append para confirmar, así que un paso rehecho encuentra su confirmación anterior en lugar de escribir dos veces. pipelines/days.mojo:

from std.os import getenv, setenv

from iceberg.batch import ColumnBuilder, batch_of
from iceberg.values import Datum
from orca import Cron, Ctx, Daily, Day, Mapped, Pipeline, Step, orca_main
from orca.tables import append, ensure_table

from lake import DAYS, days_schema

def count(mut ctx: Ctx, d: Day) raises -> Int64:
    var schema = days_schema()
    var table = ensure_table(ctx, DAYS, schema)
    var day = ColumnBuilder.of(schema, 1)       # por id de campo
    var events = ColumnBuilder.of(schema, 2)
    day.add(Datum.string_(String(d)))
    events.add(Datum.long_(Int64(d.day * 10)))
    var id = append(ctx, table, [batch_of([day^, events^])])
    ctx.log(String("instantánea añadida ", id))
    return id

comptime Count = Step["count", Day, Int64, count, Mapped, writes=[DAYS]]
comptime Days = Pipeline[
    "days", Count, partitions=Daily["2024-01-01"], schedule=Cron["0 3 * * *"]
]

def main() raises:
    # En AWS la pila establece ORCA_WAREHOUSE; localmente, por defecto un directorio.
    if not getenv("ORCA_WAREHOUSE"):
        _ = setenv("ORCA_WAREHOUSE", "build/warehouse")
    orca_main[Days]()

• Poner por defecto ORCA_WAREHOUSE en main es lo que hace el ejemplo de taxi de orca; nunca sobrescribas un valor que ya está establecido.
• Entrada Parquet ya viene como RecordBatches; pásalos a append. scan(ctx, table) lee una entrada declarada en una instantánea fijada. references/lake.md tiene particiones, escaneos, filtros y columnas anidadas.

Una app flare sobre el lago

Una app es un Router de flare servido en PORT. bin/console.mojo:

from std.os import getenv

from flare.http import HttpServer, Request, Response, Router, ServerConfig
from flare.http import html_escape, internal_error, ok_json
from flare.net import SocketAddr

from lake import load

def _html(var body: String) raises -> Response:
    var resp = Response(status=200, reason="OK", body=List[UInt8](body.as_bytes()))
    resp.headers.set("Content-Type", "text/html; charset=utf-8")
    return resp^

def days(req: Request) raises -> Response:
    try:
        var rows = load("demo", "days").scan().select(["day", "events"]).to_table()
        var h = String("<!doctype html><title>Días</title><h1>Días</h1><table>")
        for r in range(rows.num_rows()):
            h.write("<tr><td>", html_escape(rows.value(r, 0).repr_()), "</td><td>")
            h.write(html_escape(rows.value(r, 1).repr_()), "</td></tr>")
        h.write("</table>")
        return _html(h^)
    except e:
        return internal_error(String(e))

def days_json(req: Request) raises -> Response:
    return ok_json(load("demo", "days").scan().to_table().to_json())

def main() raises:
    var r = Router()
    r.get("/", days)
    r.get("/days.json", days_json)
    var addr = String("0.0.0.0:", getenv("PORT", "8080"))
    var server = HttpServer.bind(SocketAddr.parse(addr), ServerConfig())
    print("consola en http://", addr, sep="")
    server.serve(r^)

• Escucha en 0.0.0.0:$PORT. En AWS, Caddy llega a la app por su nombre de contenedor en ese puerto; 127.0.0.1 es inalcanzable desde él.

• Cada solicitud escanea la tabla nuevamente (desde S3, en un bucle de eventos): bien para una consola pequeña. Para estado, escribe una estructura que conforme a Handler, como hace la interfaz de usuario de orca (struct Ui en src/orca/ui.mojo, en el repositorio de orca).

• Parámetros de ruta: r.get("/repos/:id", repo), luego req.param("id").

• Descargar con el HttpClient de flare: un Response no puede ceder su cuerpo (return resp.body^ falla con "destroyed out of the middle of a value"). Intercámbialo; swap es una función incorporada, sin importación:

  from flare.http import HttpClient, Request
  
  var c = HttpClient(timeout_ms=120_000)
  var resp = c.send(Request(method="GET", url=url))
  var body = List[UInt8]()
  swap(body, resp.body)
  return body^
  

• En AWS una app solo recibe AWS_REGION, ORCA_WAREHOUSE, ORCA_PUBLIC_WAREHOUSE y PORT, y lee S3 con el rol de la instancia de control. Lee el lago, nunca la base de datos de orca.

Ejecutarlo localmente

mkdir -p build
mojo build bin/scheduler.mojo -o build/orca
mojo build -I src pipelines/days.mojo -o build/days
mojo build -I src bin/console.mojo -o build/console
./build/days run --partitions 2024-01-01..2024-01-03   # no se necesita base de datos
PORT=8080 ./build/console                              # http://127.0.0.1:8080

Ejecútalos en pixi shell o con pixi run. El programador y la interfaz de usuario necesitan Postgres: el scripts/with-postgres.sh de orca (cópialo; el tin no lo incluye) ejecuta un comando con un servidor desechable y ORCA_DATABASE_URL establecido:

export ORCA_WAREHOUSE=$PWD/build/warehouse
scripts/with-postgres.sh sh -c '
  ./build/orca locate build/days --root . &&
  ./build/orca register ./build/days &&
  ./build/orca backfill days 2024-01-04..2024-01-05 &&
  (./build/orca ui & ./build/orca scheduler)'   # UI en http://127.0.0.1:8270

references/local.md: una base de datos persistente, inicios de sesión, MinIO.

Desplegar en AWS

orca aws está en orca-mojo 0.3.0 y posteriores; ./build/orca aws imprime su uso.

1. Inicia sesión con aws login (o establece AWS_PROFILE); docker debe ejecutarse. Confirma la Región del proyecto en AWS Settings > View all projects > Overview > Additional Info > Region. La Zona de Disponibilidad de la pila por defecto es us-east-2a; en otra Región pásale una de sus zonas a up con --zone, por ejemplo --zone eu-north-1a.
2. Crea la pila: ./build/orca aws up --email [email protected] --hostname orca.ejemplo.com --console-hostname app.ejemplo.com. El correo recibe las alertas de Budgets (--budget N, por defecto 100 dólares al mes). Sin las opciones de hostname, la pila usa las de magmalake (orca-demo.mlake.dev, libs.mlake.dev).
3. Apunta DNS a ella: ./build/orca aws status imprime PublicAddress; crea un registro A para cada hostname hacia ella. En Cloudflare, establece los registros como "Solo DNS", o Caddy no podrá obtener certificados. La interfaz de usuario responde en SslipUrl de inmediato.
4. Confirma, luego sube la imagen: ./build/orca aws push construye bin/ y pipelines/ para linux-aarch64 y lo sube al repositorio ECR de la pila, etiquetado con los primeros 12 caracteres del commit y latest.
5. Registra pipelines: ./build/orca aws register --workers 0..4 days mueve el programador y la interfaz de usuario a la nueva imagen y registra cada pipelines/<nombre> nombrado, con las banderas de orca register (--head-type, --worker-type, --workers, --spot, --secret).
6. Inicia la app: ./build/orca aws app console sirve bin/console en https://<console hostname>. Otras apps necesitan --host y un registro DNS propio.

Después de un cambio de código: confirma, push, register de nuevo (que mueve la instancia a la nueva etiqueta), y app console de nuevo. Secretos van en Secrets Manager bajo orca/; las ejecuciones los obtienen como variables, enmascarados en cada registro que orca escribe:

aws secretsmanager create-secret --name orca/github-token --secret-string "$TOKEN"
./build/orca aws register --secret GITHUB_TOKEN=orca/github-token days

Destruir con ./build/orca aws down --yes, después de vaciar los buckets. references/aws.md tiene el contenido de la pila, inicios de sesión para la interfaz de usuario, backfills en AWS y solución de problemas.

Costo

• La instancia de control se ejecuta todo el mes (t4g.large bajo demanda, 40 GB, una IP elástica), ejecute o no algo.
• Las ejecuciones inician sus propias instancias bajo demanda. Algunos planes de AWS niegan Spot, así que --spot solo funciona donde está permitido. --workers 0..N mantiene una ejecución inactiva sin workers; mide una partición antes de un backfill grande.
• Las descargas públicas se facturan al proyecto. Si cosas que funcionaban empiezan a fallar con Access Denied, pregunta si un límite de gasto pausó el proyecto (AWS Settings > Billing).

Errores comunes

Error	Qué sucede	Solución
El archivo del programador es bin/orca.mojo	oculta el paquete orca y nada se construye	nómbralo bin/scheduler.mojo
Sin plataforma linux-aarch64	orca aws push se niega	pixi workspace platform add linux-aarch64
Construir sin -I src	from lake import ... no se encuentra	pasa -I src, como hace la construcción de la imagen
Un paso escribe un conjunto de datos no en writes=	orca.tables lo rechaza	decláralo en el Step
Un paso escribe directamente con iceberg	un reintento confirma los datos dos veces	usa orca.tables.append o commit
La app escucha en 127.0.0.1	Caddy no recibe respuesta en AWS	0.0.0.0:$PORT
Subir cambios sin confirmar	la imagen los contiene pero está etiquetada y vinculada como el último commit	confirma primero
splitlines() en texto separado por tabulaciones, como git log --numstat	Mojo 1.2.0 también divide en \t: "1\t2\tf\n3\t4\tb".splitlines() tiene 6 partes, así que las filas silenciosamente no coinciden	split("\n")
b.num_rows() en un RecordBatch	'Int' does not implement the '__call__' method	RecordBatch.num_rows es un campo; ScanResult.num_rows() (congelado...)

…

[truncado para traducción]