magmalake/magmalake-apps

MagmaLake 앱은 세 가지 종류의 프로그램을 모두 포함하는 하나의 Pixi 프로젝트입니다. 모든 프로그램은 Mojo로 작성됩니다: Orca가 스케줄링하고 Iceberg 테이블을 쓰는 파이프라인, 프로젝트 자체의 orca (스케줄러 및 UI), 그리고 앱: 테이블을 읽는 콘솔과 같은 장기 실행 Flare 서버. AWS에서는 하나의 CloudFormation 스택이 하나의 컨트롤 인스턴스에서 orca, Postgres, 앱 및 Caddy를 실행하고, 각 파이프라인 실행은 자체 인스턴스에서 실행됩니다.

파이프라인에 대한 심층 정보(파티션, keyed, 센서, dista 태스크)는 orca 스킬을 참조하십시오; Mojo 코드 작성은 Modular의 mojo-syntax 스킬을 따르십시오. 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는 플랫폼에 반드시 포함되어야 합니다. 그렇지 않으면 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>이 됩니다.

Lake: 웨어하우스의 테이블

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 자체의 버킷은 비공개로 유지되며 로그와 실행 상태만 저장합니다. 누구나 서명 없이 Lake를 읽을 수 있습니다: 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 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("스냅샷 추가됨 ", 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를 참조하십시오.

Lake 위의 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에서, 단일 이벤트 루프에서): 소규모 콘솔에는 적합합니다. 상태를 유지하려면 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은 내장 함수로, 임포트가 필요 없습니다:

  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를 읽습니다. Lake는 읽지만 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달러). 호스트 이름 옵션 없이 스택은 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용으로 빌드하고 스택의 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://<콘솔 호스트 이름>에서 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 온디맨드, 40GB, 탄력적 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에서 수신 대기	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()는 (계속)

...(잘림)