magmalake/magmalake-apps

MagmaLakeアプリとは、すべてMojoで書かれた3種類のプログラムを保持する単一のpixiプロジェクトです：パイプライン（orcaがスケジュールし、Icebergテーブルに書き込む）、プロジェクト独自の**orca（スケジューラ兼UI）、およびアプリ**（テーブルを読み取るコンソールなどの長時間稼働するFlareサーバー）。AWSでは、1つのCloudFormationスタックが、1つのコントロールインスタンス上で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>を使用します。これには数分かかり、各ティンのビルドログが出力されます。ハングすることはありません。
• linux-aarch64 がプラットフォームに含まれている必要があります。そうしないとorca aws pushがビルドを拒否します。イメージはpixi install --lockedでインストールされます。
• ティンのソースを読む： 環境には、ほとんどのティン（Flareとjsonは例外）に対してコンパイル済みの.mojocパッケージのみが保持されています。そのため、ソースはGitHub（https://github.com/magmalake/<リポジトリ>、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：以下の2行
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 に配置し、パイプラインとアプリが名前とスキーマで一致するようにします。以下のサンプルはそこから3つのものをインポートしています。完全版は references/lake.md にコピーされています：

• DAYS = "demo.days"、データセットの名前。
• days_schema()、そのIcebergスキーマ（day 文字列、フィールド1; events long、フィールド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のタクシー例が行っていることです。すでに設定されている値を決して上書きしないでください。
• 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 はそこから到達できません。

• すべてのリクエストでテーブルを再度スキャンします（S3から、1つのイベントループ上で）。小さなコンソールには問題ありません。状態が必要な場合は、orcaのUIが行うように、Handler に準拠する構造体を記述します（orcaのリポジトリ の src/orca/ui.mojo の struct Ui）。

• パスパラメータ： r.get("/repos/:id", repo)、その後 req.param("id")。

• flareの HttpClient を使用したダウンロード： Response はそのボディを放棄できません（return resp.body^ は「値の途中で破棄されました」で失敗します）。代わりに swap を使用して交換します。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（コピーしてください。ティンには含まれていません）は、使い捨てサーバーと設定された 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 at 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 でゾーンの1つを渡します（例：--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のみ」に設定します。そうしないとCaddyが証明書を取得できません。UIは即座に SslipUrl で応答します。
4. コミットして、イメージをプッシュ： ./build/orca aws push は bin/ と pipelines/ をlinux-aarch64用にビルドし、コミットの最初の12文字と latest でタグ付けされたスタックのECRリポジトリにプッシュします。
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 は bin/console を https://<console hostname> で提供します。他のアプリには --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、Elastic IP）、何かを実行するかどうかに関係なく。
• 実行は独自のオンデマンドインスタンスを起動します。 一部のAWSプランはスポットを拒否するため、--spot は許可されている場所でのみ機能します。--workers 0..N は、アイドル状態の実行をワーカーなしに保ちます。大規模なバックフィルの前に1つのパーティションを測定してください。
• パブリックダウンロードはプロジェクトに請求されます。 動作していたものが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を直接使用して書き込む	再試行時にデータが2回コミットされる	orca.tables.append または commit を使用する
アプリが 127.0.0.1 でリッスンする	AWSでCaddyが応答を受け取れない	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() はメソッドです。