magmalake/magmalake-apps

magmalake应用是一个pixi项目，包含三种程序，全部使用Mojo编写：管道由Orca调度并写入Iceberg表，项目自身的**orca（调度器和UI），以及应用**：长期运行的Flare服务器，例如读取表的控制台。在AWS上，一个CloudFormation栈运行orca、Postgres、应用和Caddy在一个控制实例上，每个管道运行在其自己的实例上。

有关管道的详细信息（分区、键控、传感器、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，因此可以一次性导入。使用pixi shelf add <tin>添加其他tin。这需要几分钟并打印每个tin的构建日志；它不会挂起。
• linux-aarch64必须位于平台列表中，否则orca aws push会拒绝构建；镜像使用pixi install --locked安装。
• **阅读tin的源代码：**环境仅包含大多数tin的已编译.mojoc包（flare和json除外），因此请在GitHub上阅读源代码（https://github.com/magmalake/<repo>，位于pixi.toml中的rev）或pixi的git缓存中：在macOS上为ls -d ~/Library/Caches/rattler/cache/git-v0/checkouts/*/<rev的前7个字符>*，在Linux上为~/.cache/rattler/cache/git-v0/checkouts/。这些技能中引用的文件路径位于这些检出中。

布局

orca aws push构建确切的以下布局：

pixi.toml, pixi.lock
bin/scheduler.mojo     项目的orca：以下两行
bin/<name>.mojo        一个应用，例如bin/console.mojo
pipelines/<name>.mojo  一个管道，以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>，管道变为/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	管道（通过orca.tables）、应用、任何未经身份验证的用户
ORCA_PUBLIC_WAREHOUSE	相同位置	想要指明“公共的那个”的代码

orca自身的存储桶保持私有，仅存储日志和运行状态。任何人都可以未经身份验证读取数据湖：aws s3 ls --no-sign-request s3://mlake-public-<account>/warehouse/，或使用DuckDB和pyiceberg的匿名模式。因此，管道写入的每个表都是公开的。

将共享表定义放在src/lake.mojo中，以便管道和应用在名称和模式上保持一致。下面的示例从中导入三个内容，完整内容复制在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("附加的快照 ", 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("控制台在 http://", addr, sep="")
    server.serve(r^)

• 监听在0.0.0.0:$PORT上。 在AWS上，Caddy通过该端口上的容器名称到达应用；127.0.0.1无法从Caddy访问。

• 每个请求都重新扫描表（来自S3，在一个事件循环上）：对于小型控制台来说没问题。对于状态，编写一个符合Handler的结构体，就像orca的UI所做的（src/orca/ui.mojo中的struct Ui，位于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；在其他区域中，使用--zone传递一个可用区给up，例如--zone eu-north-1a。
2. 创建栈： ./build/orca aws up --email [email protected] --hostname orca.example.com --console-hostname app.example.com。该电子邮件将收到预算警报（--budget N，默认为每月100美元）。如果没有主机名选项，栈将使用magmalake自身的主机名（orca-demo.mlake.dev、libs.mlake.dev）。
3. 将DNS指向它： ./build/orca aws status打印PublicAddress；为每个主机名创建一个指向它的A记录。在Cloudflare上，将记录设置为“仅DNS”（DNS only），否则Caddy无法获取证书。UI立即在SslipUrl处响应。
4. 提交，然后推送镜像： ./build/orca aws push为linux-aarch64构建bin/和pipelines/，并将其推送到栈的ECR仓库，标记为提交的前12个字符和latest。
5. 注册管道： ./build/orca aws register --workers 0..4 days将调度器和UI移动到新镜像，并注册每个名为pipelines/<name>的管道，使用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在空闲运行时保持零 worker；在进行大规模回填之前测量一个分区。
• 公共下载计入项目账单。 如果之前正常工作的内容开始失败并显示“拒绝访问”（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
推送未提交的更改	镜像包含这些更改，但被标记并链接为上次提交	先提交
对制表符分隔的文本（例如git log --numstat）使用splitlines()	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()是一个函数

（文件在此处截断）