magmalake/magmalake-apps

Um aplicativo magmalake é um projeto pixi que contém três tipos de programa, todos em Mojo: pipelines que o orca agenda e que escrevem tabelas Iceberg, o próprio orca do projeto (agendador e interface) e apps: servidores flare de longa duração, como um console que lê as tabelas. Na AWS, uma pilha CloudFormation executa orca, Postgres, os apps e Caddy em uma única instância de controle, e cada execução de pipeline em instâncias próprias.

Para pipelines em profundidade (partições, chaveados, sensores, tarefas dista), carregue a habilidade orca; escreva Mojo com a habilidade mojo-syntax da Modular. O mojoshelf tem seu próprio: npx skills add mojoshelf/mojoshelf --skill mojoshelf-consume --yes.

Criar o projeto

pixi init lake -c https://conda.modular.com/max -c conda-forge
cd lake
# Em pixi.toml, em [workspace], adicione:   preview = ["pixi-build"]
pixi add "mojo-compiler==1.2.0"
pixi workspace platform add linux-aarch64   # a plataforma da imagem AWS
pixi shelf add orca-mojo
pixi add postgresql                          # um Postgres local, para desenvolvimento

• pixi shelf add orca-mojo (orca 0.3.0) adiciona orca e tudo que precisa ao pixi.toml, incluindo dista-mojo, iceberg-mojo, flare_mlake, objectstore-mojo e json_mlake, então esses importam de uma vez. Adicione outras latas com pixi shelf add <tin>. Leva minutos e imprime o log de construção de cada lata; não travou.
• linux-aarch64 deve estar nas plataformas, ou orca aws push se recusa a construir; a imagem instala com pixi install --locked.
• Lendo o código-fonte de uma lata: o ambiente contém apenas pacotes .mojoc compilados para a maioria das latas (flare e json são as exceções), então leia o código-fonte no GitHub (https://github.com/magmalake/<repositorio> no rev em pixi.toml) ou no cache git do pixi: ls -d ~/Library/Caches/rattler/cache/git-v0/checkouts/*/<primeiros 7 do rev>* no macOS, ~/.cache/rattler/cache/git-v0/checkouts/ no Linux. Caminhos de arquivo citados nestas habilidades estão nesses checkouts.

Layout

orca aws push constrói exatamente este layout:

pixi.toml, pixi.lock
bin/scheduler.mojo     o orca do projeto: as duas linhas abaixo
bin/<nome>.mojo        um app, ex.: bin/console.mojo
pipelines/<nome>.mojo  um pipeline, terminando em orca_main[...]()
src/                   código compartilhado, no caminho de import (-I src)
Dockerfile.orca        opcional: substitui o Dockerfile modelo do orca

bin/scheduler.mojo é exatamente:

from orca.command import orca_command

def main() raises:
    orca_command()

Nunca o nomeie orca.mojo: isso sombrearia o pacote orca. Na imagem ele se torna /opt/orca/bin/orca, outros bin/<nome>.mojo se tornam /opt/orca/bin/<nome>, e pipelines /opt/orca/pipelines/<nome>.

O lago: tabelas em um warehouse

O orca encontra tabelas através de ORCA_WAREHOUSE, a raiz de um catálogo Iceberg no sistema de arquivos: dataset demo.days é a tabela em <warehouse>/demo/days. O warehouse é um diretório local ou uma localização S3 (s3://bucket/warehouse). Na AWS a pilha o coloca no bucket público:

Variável	Valor na AWS	Quem lê
ORCA_WAREHOUSE	s3://mlake-public-<conta>/warehouse	pipelines (via orca.tables), apps, qualquer um sem assinatura
ORCA_PUBLIC_WAREHOUSE	a mesma localização	código que quer dizer "o público"

O bucket próprio do orca permanece privado e contém apenas logs e estado de execução. Qualquer um pode ler o lago, sem assinatura: aws s3 ls --no-sign-request s3://mlake-public-<conta>/warehouse/, ou DuckDB e pyiceberg em seus modos anônimos. Então toda tabela que um pipeline escreve é pública.

Coloque definições de tabelas compartilhadas em src/lake.mojo, para que pipelines e apps concordem com nomes e esquemas. As amostras abaixo importam três coisas dele, copiadas na íntegra em references/lake.md:

• DAYS = "demo.days", o nome do dataset;
• days_schema(), seu esquema Iceberg (day string, campo 1; events long, campo 2), de Schema.parse do JSON do Iceberg;
• load(namespace, name), uma tabela de ORCA_WAREHOUSE (padrão build/warehouse), através de FilesystemCatalog.local(root) ou, para s3://, FilesystemCatalog(root, FileIO()).

Escrevendo tabelas a partir de uma etapa

Uma etapa declara os datasets que escreve (writes=[...]) e lê (reads=[...]); orca.tables recusa qualquer outro. Use ensure_table para criar uma tabela no primeiro uso e append para commit, para que uma etapa repetida encontre seu commit anterior em vez de escrever duas vezes. 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)       # pelo id do 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("snapshot anexado ", 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:
    # Na AWS a pilha define ORCA_WAREHOUSE; localmente, padrão para um diretório.
    if not getenv("ORCA_WAREHOUSE"):
        _ = setenv("ORCA_WAREHOUSE", "build/warehouse")
    orca_main[Days]()

• Definir ORCA_WAREHOUSE como padrão em main é o que o exemplo de táxi do orca faz; nunca sobrescreva um valor que já está definido.
• Entrada Parquet já vem como RecordBatches; passe-os para append. scan(ctx, table) lê uma entrada declarada em um snapshot fixado. references/lake.md tem partições, scans, filtros e colunas aninhadas.

Um aplicativo flare sobre o lago

Um app é um Router do flare servido em 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 em http://", addr, sep="")
    server.serve(r^)

• Escute em 0.0.0.0:$PORT. Na AWS, o Caddy alcança o app pelo seu nome de container nessa porta; 127.0.0.1 é inalcançável para ele.

• Cada requisição escaneia a tabela novamente (do S3, em um loop de evento): bom para um console pequeno. Para estado, escreva uma struct conforme Handler, como a interface do orca faz (struct Ui em src/orca/ui.mojo, no repositório do orca).

• Parâmetros de caminho: r.get("/repos/:id", repo), então req.param("id").

• Download com o HttpClient do flare: um Response não pode ceder seu corpo (return resp.body^ falha com "destroyed out of the middle of a value"). Troque-o; swap é embutido, sem 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^
  

• Na AWS um app recebe apenas AWS_REGION, ORCA_WAREHOUSE, ORCA_PUBLIC_WAREHOUSE e PORT, e lê S3 com a role da instância de controle. Ele lê o lago, nunca o banco de dados do orca.

Executar 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   # nenhum banco de dados necessário
PORT=8080 ./build/console                              # http://127.0.0.1:8080

Execute-os em pixi shell ou com pixi run. O scheduler e a interface precisam de Postgres: o scripts/with-postgres.sh do orca (copie-o; a lata não o envia) executa um comando com um servidor descartável e ORCA_DATABASE_URL definido:

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

references/local.md: um banco de dados persistente, logins, MinIO.

Implantar na AWS

orca aws está no orca-mojo 0.3.0 e posteriores; ./build/orca aws imprime seu uso.

1. Faça login com aws login (ou defina AWS_PROFILE); docker deve estar em execução. Confirme a Região do projeto em AWS Settings > View all projects > Overview > Additional Info > Region. A Zona de Disponibilidade da pilha padrão é us-east-2a; em outra Região, passe uma de suas zonas para up com --zone, como --zone eu-north-1a.
2. Crie a pilha: ./build/orca aws up --email [email protected] --hostname orca.exemplo.com --console-hostname app.exemplo.com. O email recebe os alertas de Budgets (--budget N, padrão 100 dólares por mês). Sem as opções de hostname, a pilha usa os do magmalake (orca-demo.mlake.dev, libs.mlake.dev).
3. Aponte o DNS para ela: ./build/orca aws status imprime PublicAddress; crie um registro A para cada hostname apontando para ele. No Cloudflare, defina os registros como "DNS only", ou o Caddy não conseguirá certificados. A UI responde em SslipUrl imediatamente.
4. Faça commit, depois envie a imagem: ./build/orca aws push constrói bin/ e pipelines/ para linux-aarch64 e a envia para o repositório ECR da pilha, marcada com os primeiros 12 caracteres do commit e latest.
5. Registre pipelines: ./build/orca aws register --workers 0..4 days move o scheduler e a UI para a nova imagem e registra cada pipelines/<nome>, com as flags do orca register (--head-type, --worker-type, --workers, --spot, --secret).
6. Inicie o app: ./build/orca aws app console serve bin/console em https://<console hostname>. Outros apps precisam de --host e um registro DNS próprio.

Após uma alteração no código: commit, push, register novamente (que move a instância para a nova tag) e app console novamente. Segredos vão no Secrets Manager sob orca/; execuções os recebem como variáveis, mascaradas em todos os logs que o orca escreve:

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

Destrua com ./build/orca aws down --yes, após esvaziar os buckets. references/aws.md tem o conteúdo da pilha, logins para a UI, backfills na AWS e solução de problemas.

Custo

• A instância de controle executa o mês inteiro (t4g.large sob demanda, 40 GB, um Elastic IP), quer algo execute ou não.
• Execuções iniciam instâncias sob demanda próprias. Alguns planos AWS negam Spot, então --spot funciona apenas onde permitido. --workers 0..N mantém uma execução ociosa sem workers; meça uma partição antes de um grande backfill.
• Downloads públicos são cobrados no projeto. Se coisas que funcionavam começarem a falhar com Access Denied, pergunte se um limite de gastos pausou o projeto (AWS Settings > Billing).

Erros comuns

Erro	O que acontece	Correção
O arquivo do scheduler é bin/orca.mojo	Ele sombreia o pacote orca e nada compila	Nomeie-o bin/scheduler.mojo
Sem a plataforma linux-aarch64	orca aws push se recusa	pixi workspace platform add linux-aarch64
Compilar sem -I src	from lake import ... não é encontrado	passe -I src, como a compilação da imagem faz
Uma etapa escreve um dataset não em writes=	orca.tables recusa	declare-o no Step
Uma etapa escreve com iceberg diretamente	uma repetição comete os dados duas vezes	use orca.tables.append ou commit
O app escuta em 127.0.0.1	Caddy não obtém resposta na AWS	0.0.0.0:$PORT
Enviar alterações não commitadas	a imagem as contém, mas é marcada e vinculada como o último commit	faça commit primeiro
splitlines() em texto separado por tabulação, como git log --numstat	Mojo 1.2.0 também divide em \t: "1\t2\tf\n3\t4\tb".splitlines() tem 6 partes, então linhas silenciosamente não correspondem	split("\n")
b.num_rows() em um RecordBatch	'Int' does not implement the '__call__' method	RecordBatch.num_rows é um campo; ScanResult.num_rows() (congelado) é um método