magmalake/magmalake-apps

一個 magmalake 應用程式是一個 pixi 專案，其中包含三種程式，全部以 Mojo 撰寫：pipeline（由 orca 排程，寫入 Iceberg 表格）、專案自己的 orca（排程器與 UI），以及應用程式：長時間執行的 flare 伺服器，例如讀取表格的主控台。在 AWS 上，一個 CloudFormation 堆疊會在同一個控制實例上執行 orca、Postgres、應用程式和 Caddy，而每個 pipeline 則在其各自的實例上執行。

若需深入了解 pipeline（分割區、鍵控、感測器、dista 任務），請載入 orca 技能；使用 Modular 的 mojo-syntax 技能來撰寫 Mojo。mojoshelf 自身：npx skills add mojoshelf/mojoshelf --skill mojoshelf-consume --yes。

建立專案

pixi init lake -c https://conda.modular.com/max -c conda-forge
cd lake
# 在 pixi.toml 的 [workspace] 中加入：   preview = ["pixi-build"]
pixi add "mojo-compiler==1.2.0"
pixi workspace platform add linux-aarch64   # AWS 映像的平台
pixi shelf add orca-mojo
pixi add postgresql                          # 本機 Postgres，用於開發

• pixi shelf add orca-mojo（orca 0.3.0）會將 orca 及其所有依賴項加入 pixi.toml，包括 dista-mojo、iceberg-mojo、flare_mlake、objectstore-mojo 和 json_mlake，因此這些套件可立即匯入。若要加入其他 tin，請使用 pixi shelf add <tin>。這需要幾分鐘，並會列印每個 tin 的建置記錄；它並未掛起。
• linux-aarch64 必須在 platforms 中，否則 orca aws push 會拒絕建置；映像會使用 pixi install --locked 安裝。
• 閱讀 tin 的原始碼： 環境中只包含大多數 tin（flare 和 json 是例外）的已編譯 .mojoc 套件，因此請在 GitHub（https://github.com/magmalake/<repo>，使用 pixi.toml 中的 rev）或 pixi 的 git 快取中閱讀原始碼： ls -d ~/Library/Caches/rattler/cache/git-v0/checkouts/*/<rev 的前 7 碼>*（macOS），~/.cache/rattler/cache/git-v0/checkouts/（Linux）。這些技能中引用的檔案路徑都在這些 checkout 中。

佈局

orca aws push 會建置完全符合此佈局的結構：

pixi.toml, pixi.lock
bin/scheduler.mojo     專案的 orca：下方兩行
bin/<name>.mojo        一個應用程式，例如 bin/console.mojo
pipelines/<name>.mojo  一個 pipeline，結尾為 orca_main[...]()
src/                   共用程式碼，位於匯入路徑上（-I src）
Dockerfile.orca        可選：取代 orca 的範本 Dockerfile

bin/scheduler.mojo 的內容為：

from orca.command import orca_command

def main() raises:
    orca_command()

切勿將其命名為 orca.mojo：它會遮蔽 orca 套件。在映像中，它會變成 /opt/orca/bin/orca，其他 bin/<name>.mojo 會變成 /opt/orca/bin/<name>，而 pipeline 會變成 /opt/orca/pipelines/<name>。

資料湖：倉儲中的表格

orca 透過 ORCA_WAREHOUSE 尋找表格，這是 Iceberg 檔案系統目錄的根目錄：資料集 demo.days 是位於 <warehouse>/demo/days 的表格。倉儲可以是本機目錄或 S3 位置（s3://bucket/warehouse）。在 AWS 上，堆疊會將其放置在公開儲存桶中：

變數	AWS 上的值	誰讀取它
ORCA_WAREHOUSE	s3://mlake-public-<account>/warehouse	pipeline（透過 orca.tables）、應用程式、任何未簽署者
ORCA_PUBLIC_WAREHOUSE	相同位置	想要指定「公開的」那個的程式碼

orca 自己的儲存桶保持私有，僅儲存日誌和運行狀態。任何人都可以未簽署狀態讀取資料湖：aws s3 ls --no-sign-request s3://mlake-public-<account>/warehouse/，或使用 DuckDB 和 pyiceberg 的匿名模式。因此，pipeline 寫入的每個表格都是公開的。

將共用的表格定義放在 src/lake.mojo 中，以便 pipeline 和應用程式在名稱和結構上保持一致。下面的範例從中共用了三項，完整內容請見 references/lake.md：

• DAYS = "demo.days"，資料集的名稱；
• days_schema()，其 Iceberg 結構（day 字串，欄位 1；events 長整數，欄位 2），來自 Iceberg JSON 的 Schema.parse；
• load(namespace, name)，從 ORCA_WAREHOUSE（預設為 build/warehouse）載入表格，透過 FilesystemCatalog.local(root) 或（若為 s3://）FilesystemCatalog(root, FileIO())。

從步驟寫入表格

一個步驟會宣告它寫入（writes=[...]）和讀取（reads=[...]）的資料集；orca.tables 會拒絕任何其他資料集。使用 ensure_table 在首次使用時建立表格，並使用 append 來提交，以便重試的步驟能找到其先前的提交，而不是重複寫入。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)       # 依欄位 ID
    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("appended snapshot ", 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:
    # 在 AWS 上，堆疊會設定 ORCA_WAREHOUSE；在本機，預設為一個目錄。
    if not getenv("ORCA_WAREHOUSE"):
        _ = setenv("ORCA_WAREHOUSE", "build/warehouse")
    orca_main[Days]()

• 在 main 中預設 ORCA_WAREHOUSE 是 orca 的 taxi 範例的做法；切勿覆蓋已設定的值。
• Parquet 輸入 已經以 RecordBatch 的形式提供；將它們傳遞給 append。scan(ctx, table) 會在固定的快照讀取宣告的輸入。references/lake.md 包含分割區、掃描、篩選和巢狀欄位。

基於資料湖的 flare 應用程式

一個應用程式是在 PORT 上服務的 flare Router。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 on http://", addr, sep="")
    server.serve(r^)

• 在 0.0.0.0:$PORT 上監聽。 在 AWS 上，Caddy 會透過容器名稱在那個連接埠上存取應用程式；127.0.0.1 無法從 Caddy 到達。

• 每次請求都重新掃描表格（來自 S3，在同一個事件迴圈中）：對於小型主控台來說沒問題。若要維護狀態，請撰寫一個符合 Handler 的結構，就像 orca 的 UI 那樣（struct Ui 在 src/orca/ui.mojo 中，位於 orca 的儲存庫）。

• 路徑參數： r.get("/repos/:id", repo)，然後使用 req.param("id")。

• 使用 flare 的 HttpClient 下載： Response 無法交出它的 body（return resp.body^ 會失敗，並顯示「destroyed out of the middle of a value」）。將其交換出來；swap 是內建函式，無需匯入：

  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^
  

• 在 AWS 上，應用程式只會取得 AWS_REGION、ORCA_WAREHOUSE、ORCA_PUBLIC_WAREHOUSE 和 PORT，並使用控制實例的角色讀取 S3。它讀取的是資料湖，而不是 orca 的資料庫。

在本機執行

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   # 不需要資料庫
PORT=8080 ./build/console                              # http://127.0.0.1:8080

在 pixi shell 或使用 pixi run 執行它們。排程器和 UI 需要 Postgres：orca 的 scripts/with-postgres.sh（複製它；tin 並未包含它）會使用一個臨時伺服器和設定的 ORCA_DATABASE_URL 來執行指令：

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：持久的資料庫、登入、MinIO。

部署到 AWS

orca aws 包含在 orca-mojo 0.3.0 及更高版本中；./build/orca aws 會列印其使用方式。

1. 登入 使用 aws login（或設定 AWS_PROFILE）；docker 必須正在執行。在 AWS 設定 > 檢視所有專案 > 概述 > 其他資訊 > 區域中確認專案的區域。堆疊的可用區域預設為 us-east-2a；若在其他區域，請將其中一個可用區域傳遞給 up 的 --zone 參數，例如 --zone eu-north-1a。
2. 建立堆疊： ./build/orca aws up --email [email protected] --hostname orca.example.com --console-hostname app.example.com。該電子郵件會收到預算提醒（--budget N，預設為每月 100 美元）。若未提供 hostname 選項，堆疊將使用 magmalake 自己的 hostname（orca-demo.mlake.dev、libs.mlake.dev）。
3. 將 DNS 指向它： ./build/orca aws status 會列印 PublicAddress；為每個 hostname 建立一個指向該位址的 A 記錄。在 Cloudflare 上，將記錄設為「僅限 DNS」，否則 Caddy 無法取得憑證。UI 會立即在 SslipUrl 回應。
4. 提交，然後推送映像： ./build/orca aws push 會為 linux-aarch64 建置 bin/ 和 pipelines/，並將其推送到堆疊的 ECR 儲存庫，標籤為提交的前 12 個字元和 latest。
5. 註冊 pipeline： ./build/orca aws register --workers 0..4 days 會將排程器和 UI 移動到新的映像，並註冊每個名為 pipelines/<name> 的 pipeline，使用 orca register 的旗標（--head-type、--worker-type、--workers、--spot、--secret）。
6. 啟動應用程式： ./build/orca aws app console 會在 https://<console hostname> 服務 bin/console。其他應用程式需要 --host 和自己專屬的 DNS 記錄。

程式碼變更後：提交、push、再次 register（這會將實例移動到新的標籤），然後再次 app console。秘密 放在 Secrets Manager 中，路徑為 orca/；執行時會將其作為變數取得，並在 orca 寫入的所有日誌中遮罩：

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

拆除 使用 ./build/orca aws down --yes，在清空儲存桶之後。references/aws.md 包含堆疊的內容、UI 的登入方式、AWS 上的回填和疑難排解。

成本

• 控制實例整月執行（t4g.large 隨需執行個體，40 GB，一個彈性 IP），無論是否有任何東西在執行。
• 執行會啟動自己專屬的隨需執行個體。 某些 AWS 方案不允許 Spot 執行個體，因此 --spot 僅在允許的地方有效。--workers 0..N 可讓閒置的執行保持零個工作者；在大型回填之前，請先測量一個分割區。
• 公開下載會向專案收費。 如果之前正常運作的功能開始失敗並顯示 Access Denied，請詢問是否花費限制已暫停專案（AWS 設定 > 帳單）。

常見錯誤

錯誤	結果	修正
排程器檔案是 bin/orca.mojo	它會遮蔽 orca 套件，且無法建置	將其命名為 bin/scheduler.mojo
沒有 linux-aarch64 平台	orca aws push 會拒絕	pixi workspace platform add linux-aarch64
建置時沒有 -I src	找不到 from lake import ...	傳遞 -I src，就像映像的建置所做的那樣
步驟寫入的資料集不在 writes= 中	orca.tables 會拒絕	在 Step 中宣告它
步驟直接使用 iceberg 寫入	重試時會提交資料兩次	使用 orca.tables.append 或 commit
應用程式在 127.0.0.1 上監聽	Caddy 在 AWS 上得不到回應	0.0.0.0:$PORT
推送未提交的變更	映像會保留這些變更，但標籤和連結是最後一次提交	先提交
對以 Tab 分隔的文字使用 splitlines()，例如 git log --numstat	Mojo 1.2.0 也會在 \t 上分割："1\t2\tf\n3\t4\tb".splitlines() 會有 6 個部分，因此行會悄悄不符	使用 split("\n")
在 RecordBatch 上使用 b.num_rows()	'Int' does not implement the '__call__' method	RecordBatch.num_rows 是一個欄位；ScanResult.num_rows() 是一個方法