magmalake/magmalake-apps

Une application magmalake est un seul projet pixi contenant trois types de programmes, tous en Mojo : les pipelines qu'orca planifie et qui écrivent des tables Iceberg, le propre orca du projet (planificateur et interface utilisateur), et les apps : des serveurs flare longue durée, comme une console qui lit les tables. Sur AWS, une stack CloudFormation exécute orca, Postgres, les apps et Caddy sur une seule instance de contrôle, et chaque exécution de pipeline sur ses propres instances.

Pour les pipelines en profondeur (partitions, clés, capteurs, tâches dista), chargez le skill orca ; écrivez du Mojo avec le skill mojo-syntax de Modular. La bibliothèque mojoshelf : npx skills add mojoshelf/mojoshelf --skill mojoshelf-consume --yes.

Créer le projet

pixi init lake -c https://conda.modular.com/max -c conda-forge
cd lake
# Dans pixi.toml, section [workspace], ajoutez :   preview = ["pixi-build"]
pixi add "mojo-compiler==1.2.0"
pixi workspace platform add linux-aarch64   # la plateforme de l'image AWS
pixi shelf add orca-mojo
pixi add postgresql                          # un Postgres local, pour le développement

• pixi shelf add orca-mojo (orca 0.3.0) ajoute orca et tout ce dont il a besoin à pixi.toml, y compris dista-mojo, iceberg-mojo, flare_mlake, objectstore-mojo et json_mlake, donc ces imports sont disponibles immédiatement. Ajoutez d'autres boîtes avec pixi shelf add <tin>. Cela prend quelques minutes et affiche le log de construction de chaque boîte ; cela n'a jamais bloqué.
• linux-aarch64 doit être dans les plateformes, sinon orca aws push refuse de construire ; l'image s'installe avec pixi install --locked.
• Lecture du code source d'une boîte : l'environnement ne contient que des paquets .mojoc compilés pour la plupart des boîtes (flare et json sont les exceptions), donc lisez la source sur GitHub (https://github.com/magmalake/<repo> à la rev dans pixi.toml) ou dans le cache git de pixi : ls -d ~/Library/Caches/rattler/cache/git-v0/checkouts/*/<7 premiers car. de rev>* sur macOS, ~/.cache/rattler/cache/git-v0/checkouts/ sur Linux. Les chemins de fichiers cités dans ces skills se trouvent dans ces répertoires de checkout.

Structure

orca aws push construit exactement cette structure :

pixi.toml, pixi.lock
bin/scheduler.mojo     l'orca du projet : les deux lignes ci-dessous
bin/<nom>.mojo        une app, par ex. bin/console.mojo
pipelines/<nom>.mojo  un pipeline, se terminant par orca_main[...]()
src/                   code partagé, sur le chemin d'import (-I src)
Dockerfile.orca        facultatif : remplace le Dockerfile modèle d'orca

bin/scheduler.mojo est exactement :

from orca.command import orca_command

def main() raises:
    orca_command()

Ne le nommez jamais orca.mojo : cela masquerait le paquet orca. Dans l'image il devient /opt/orca/bin/orca, les autres bin/<nom>.mojo deviennent /opt/orca/bin/<nom>, et les pipelines /opt/orca/pipelines/<nom>.

Le lac : tables dans un entrepôt

orca trouve les tables via ORCA_WAREHOUSE, la racine d'un catalogue de système de fichiers Iceberg : le jeu de données demo.days est la table à <warehouse>/demo/days. L'entrepôt est un répertoire local ou une localisation S3 (s3://bucket/warehouse). Sur AWS, la stack le place dans le bucket public :

Variable	Valeur sur AWS	Qui la lit
ORCA_WAREHOUSE	s3://mlake-public-<account>/warehouse	pipelines (via orca.tables), apps, toute personne non authentifiée
ORCA_PUBLIC_WAREHOUSE	la même localisation	code qui veut dire "le public"

Le propre bucket d'orca reste privé et ne contient que des logs et l'état d'exécution. N'importe qui peut lire le lac, sans authentification : aws s3 ls --no-sign-request s3://mlake-public-<account>/warehouse/, ou DuckDB et pyiceberg en mode anonyme. Ainsi, chaque table qu'un pipeline écrit est publique.

Mettez les définitions de tables partagées dans src/lake.mojo, pour que les pipelines et les apps soient d'accord sur les noms et les schémas. Les exemples ci-dessous importent trois choses de ce fichier, copiées intégralement dans references/lake.md :

• DAYS = "demo.days", le nom du jeu de données ;
• days_schema(), son schéma Iceberg (day string, champ 1 ; events long, champ 2), à partir de Schema.parse du JSON d'Iceberg ;
• load(namespace, name), une table depuis ORCA_WAREHOUSE (par défaut build/warehouse), via FilesystemCatalog.local(root) ou, pour s3://, FilesystemCatalog(root, FileIO()).

Écrire des tables depuis une étape

Une étape déclare les jeux de données qu'elle écrit (writes=[...]) et lit (reads=[...]) ; orca.tables en refuse tout autre. Utilisez ensure_table pour créer une table lors de la première utilisation et append pour valider, afin qu'une étape relancée trouve sa validation précédente au lieu d'écrire deux fois. 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)       # par id de champ
    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("snapshot ajouté ", 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:
    # Sur AWS, la stack définit ORCA_WAREHOUSE ; localement, valeur par défaut un répertoire.
    if not getenv("ORCA_WAREHOUSE"):
        _ = setenv("ORCA_WAREHOUSE", "build/warehouse")
    orca_main[Days]()

• Définir ORCA_WAREHOUSE par défaut dans main est ce que fait l'exemple taxi d'orca ; ne jamais remplacer une valeur déjà définie.
• L'entrée Parquet arrive déjà sous forme de RecordBatch ; passez-les à append. scan(ctx, table) lit une entrée déclarée à un snapshot figé. references/lake.md contient les partitions, scans, filtres et colonnes imbriquées.

Une application flare sur le lac

Une application est un Router flare servi sur 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>Days</title><h1>Days</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("console sur http://", addr, sep="")
    server.serve(r^)

• Écouter sur 0.0.0.0:$PORT. Sur AWS, Caddy atteint l'application par son nom de conteneur sur ce port ; 127.0.0.1 est inaccessible depuis Caddy.

• Chaque requête scanne à nouveau la table (depuis S3, sur une boucle d'événements) : acceptable pour une petite console. Pour un état, écrivez une struct conforme à Handler, comme fait l'interface utilisateur d'orca (struct Ui dans src/orca/ui.mojo, dans le dépôt d'orca).

• Paramètres de chemin : r.get("/repos/:id", repo), puis req.param("id").

• Téléchargement avec HttpClient de flare : une Response ne peut pas céder son corps (return resp.body^ échoue avec "destroyed out of the middle of a value"). Échangez-le ; swap est une fonction intégrée, pas d'import :

  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^
  

• Sur AWS, une application ne reçoit que AWS_REGION, ORCA_WAREHOUSE, ORCA_PUBLIC_WAREHOUSE et PORT, et lit S3 avec le rôle de l'instance de contrôle. Elle lit le lac, jamais la base de données d'orca.

Exécution locale

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   # pas de base de données nécessaire
PORT=8080 ./build/console                              # http://127.0.0.1:8080

Exécutez-les dans pixi shell ou avec pixi run. Le planificateur et l'interface utilisateur ont besoin de Postgres : scripts/with-postgres.sh d'orca (copiez-le ; la boîte ne le fournit pas) exécute une commande avec un serveur jetable et ORCA_DATABASE_URL défini :

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 à http://127.0.0.1:8270

references/local.md : une base de données persistante, connexions, MinIO.

Déploiement sur AWS

orca aws est disponible dans orca-mojo 0.3.0 et ultérieur ; ./build/orca aws affiche son utilisation.

1. Connectez-vous avec aws login (ou définissez AWS_PROFILE) ; docker doit être en cours d'exécution. Confirmez la Région du projet dans Paramètres AWS > Afficher tous les projets > Aperçu > Informations supplémentaires > Région. La zone de disponibilité de la stack par défaut est us-east-2a ; dans une autre Région, passez une de ses zones à up avec --zone, par exemple --zone eu-north-1a.
2. Créez la stack : ./build/orca aws up --email [email protected] --hostname orca.example.com --console-hostname app.example.com. L'email reçoit les alertes Budgets (--budget N, par défaut 100 dollars par mois). Sans les options de nom d'hôte, la stack utilise ceux de magmalake (orca-demo.mlake.dev, libs.mlake.dev).
3. Pointez le DNS vers la stack : ./build/orca aws status affiche PublicAddress ; créez un enregistrement A pour chaque nom d'hôte pointant vers cette adresse. Sur Cloudflare, définissez les enregistrements sur "DNS uniquement", sinon Caddy ne peut pas obtenir de certificats. L'interface utilisateur répond immédiatement à SslipUrl.
4. Validez, puis poussez l'image : ./build/orca aws push construit bin/ et pipelines/ pour linux-aarch64 et la pousse vers le dépôt ECR de la stack, étiquetée avec les 12 premiers caractères du commit et latest.
5. Enregistrez les pipelines : ./build/orca aws register --workers 0..4 days déplace le planificateur et l'interface utilisateur vers la nouvelle image et enregistre chaque pipelines/<nom> nommé, avec les indicateurs d'orca register (--head-type, --worker-type, --workers, --spot, --secret).
6. Démarrez l'application : ./build/orca aws app console sert bin/console à https://<nom d'hôte de la console>. Les autres applications ont besoin de --host et d'un enregistrement DNS propre.

Après une modification du code : validez, push, register à nouveau (qui déplace l'instance vers le nouveau tag), et app console à nouveau. Les secrets vont dans Secrets Manager sous orca/ ; les exécutions les reçoivent comme variables, masquées dans chaque log qu'orca écrit :

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

Démontez avec ./build/orca aws down --yes, après avoir vidé les buckets. references/aws.md contient le contenu de la stack, les connexions pour l'interface utilisateur, les backfills sur AWS et le dépannage.

Coût

• L'instance de contrôle tourne tout le mois (t4g.large à la demande, 40 Go, une IP élastique), que quelque chose s'exécute ou non.
• Les exécutions démarrent des instances à la demande dédiées. Certains plans AWS refusent Spot, donc --spot fonctionne uniquement là où c'est autorisé. --workers 0..N maintient une exécution inactive sans workers ; mesurez une partition avant un backfill important.
• Les téléchargements publics sont facturés au projet. Si des choses qui fonctionnaient commencent à échouer avec Accès Refusé, demandez si une limite de dépenses a mis en pause le projet (Paramètres AWS > Facturation).

Erreurs courantes

Erreur	Ce qui se produit	Correctif
Le fichier du planificateur est bin/orca.mojo	il masque le paquet orca et rien ne se construit	nommez-le bin/scheduler.mojo
Pas de plateforme linux-aarch64	orca aws push refuse	pixi workspace platform add linux-aarch64
Construction sans -I src	from lake import ... n'est pas trouvé	passez -I src, comme le fait la construction de l'image
Une étape écrit un jeu de données pas dans writes=	orca.tables le refuse	déclarez-le dans le Step
Une étape écrit avec iceberg directement	une nouvelle tentative valide les données deux fois	utilisez orca.tables.append ou commit
L'application écoute sur 127.0.0.1	Caddy ne reçoit aucune réponse sur AWS	0.0.0.0:$PORT
Pousser des modifications non validées	l'image les contient mais est étiquetée et liée comme le dernier commit	validez d'abord
splitlines() sur du texte séparé par des tabulations, par ex. git log --numstat	Mojo 1.2.0 divise aussi sur \t : "1\t2\tf\n3\t4\tb".splitlines() donne 6 parties, donc les lignes ne correspondent pas silencieusement	split("\n")
b.num_rows() sur un RecordBatch	'Int' n'implémente pas la méthode '__call__'	RecordBatch.num_rows est un champ ; ScanResult.num_rows() (à partir de

…

[suite tronquée pour la traduction]