diff --git a/plugins/flyte/skills/flyte-sdk-app/SKILL.md b/plugins/flyte/skills/flyte-sdk-app/SKILL.md index 43d73d3..bf1aed7 100644 --- a/plugins/flyte/skills/flyte-sdk-app/SKILL.md +++ b/plugins/flyte/skills/flyte-sdk-app/SKILL.md @@ -27,8 +27,12 @@ Build and serve applications with Flyte 2. | `StreamlitAppEnvironment` | Dashboards, data apps | `from flyte.app.extras import StreamlitAppEnvironment` | | `vLLMAppEnvironment` | LLM serving | `from flyte.app.extras import vLLMAppEnvironment` | | `SGLangAppEnvironment` | Structured generation | `from flyte.app.extras import SGLangAppEnvironment` | +| `FlyteWebhookAppEnvironment` | Pre-built control-plane webhook | `from flyte.app.extras import FlyteWebhookAppEnvironment` | +| Ollama (no plugin) | Lightweight local-style LLM serving | generic `flyte.app.AppEnvironment` + entrypoint | | Custom (`AppEnvironment`) | Any HTTP server | `import flyte` | +> [!NOTE] **Ollama has no dedicated `*AppEnvironment`.** vLLM and SGLang ship plugins; Ollama is served with the generic `AppEnvironment` plus a small entrypoint that launches `ollama serve` and pulls the model on startup. + ## FastAPI App — Model Serving ### Basic FastAPI app @@ -459,6 +463,68 @@ if __name__ == "__main__": flyte.serve(env) ``` +## Ollama App + +Ollama serves open models locally with an OpenAI-compatible API. Because there is no dedicated plugin, build a generic `AppEnvironment` whose image installs Ollama and whose `args` run a `--server` entrypoint that starts `ollama serve` and pulls the model on startup: + +```python +import os +import subprocess +import sys +import flyte + +MODEL = "llama3.2" +PORT = 11434 + +image = ( + flyte.Image.from_debian_base() + .with_commands(["curl -fsSL https://ollama.com/install.sh | sh"]) +) + +ollama_app = flyte.app.AppEnvironment( + name="ollama-app", + image=image, + args=["python", __file__, "--server"], + port=PORT, + resources=flyte.Resources(cpu="4", memory="16Gi", gpu="L40s:1", disk="20Gi"), # GPU optional + scaling=flyte.app.Scaling(replicas=(0, 1), scaledown_after=300), + requires_auth=False, +) + +def serve() -> None: + server = subprocess.Popen(["ollama", "serve"], env={**os.environ, "OLLAMA_HOST": f"0.0.0.0:{PORT}"}) + subprocess.run(["ollama", "pull", MODEL], check=True) + server.wait() + +if __name__ == "__main__": + if "--server" in sys.argv: + serve() + else: + flyte.init_from_config() + flyte.serve(ollama_app) +``` + +## Native Webhook App (FlyteWebhookAppEnvironment) + +`FlyteWebhookAppEnvironment` is a pre-built `FastAPIAppEnvironment` that exposes HTTP endpoints for common Flyte control-plane operations — no hand-written routes needed. Endpoints: `core` (`/health`, `/me`), `task` (`POST /run-task/...`), `run` (`GET /run/{name}`, `POST /run/{name}/abort`), `app` (activate/deactivate/call), `trigger`, `build`, and `prefetch`. Select the surface you need: + +```python +import flyte +import flyte.app +from flyte.app.extras import FlyteWebhookAppEnvironment + +webhook_env = FlyteWebhookAppEnvironment( + name="my-webhook", + title="My Flyte Webhook", + endpoint_groups=["core", "task", "run"], # or `endpoints=[...]` for individual routes + task_allowlist=["prod/my-project/allowed-task"], # optional restrictive allowlist + resources=flyte.Resources(cpu=1, memory="512Mi"), + requires_auth=True, +) +``` + +For a bespoke webhook that triggers tasks with your own auth and validation, use `FastAPIAppEnvironment` directly — see the **Webhook Apps** section above. + ## App Secrets ### Secret-based authentication diff --git a/plugins/flyte/skills/flyte-sdk-author/SKILL.md b/plugins/flyte/skills/flyte-sdk-author/SKILL.md index ad1173e..1cebd26 100644 --- a/plugins/flyte/skills/flyte-sdk-author/SKILL.md +++ b/plugins/flyte/skills/flyte-sdk-author/SKILL.md @@ -48,13 +48,13 @@ env = flyte.TaskEnvironment( async def preprocess(data: list[str]) -> flyte.io.File: # ETL: clean + write to remote storage ... - return flyte.io.File(path="/tmp/output.parquet") + return await flyte.io.File.from_local("/tmp/output.parquet") @env.task async def train(data_path: flyte.io.File) -> flyte.io.File: # ML: model training ... - return flyte.io.File(path="/tmp/model.pt") + return await flyte.io.File.from_local("/tmp/model.pt") @env.task async def evaluate(model_path: flyte.io.File) -> dict: @@ -78,7 +78,7 @@ if __name__ == "__main__": ```python @env.task async def process_image(url: str) -> flyte.io.File: - return flyte.io.File(path=f"/tmp/{url.split('/')[-1]}.png") + return await flyte.io.File.from_local(f"/tmp/{url.split('/')[-1]}.png") @env.task async def main(urls: list[str]) -> list: @@ -270,7 +270,7 @@ async def train( ) -> flyte.io.File: """Train a model and save checkpoint.""" ... - return flyte.io.File(path="/tmp/checkpoint.pt") + return await flyte.io.File.from_local("/tmp/checkpoint.pt") @env.task async def evaluate( diff --git a/plugins/flyte/skills/flyte-sdk-data/SKILL.md b/plugins/flyte/skills/flyte-sdk-data/SKILL.md index 57e269e..45a6daa 100644 --- a/plugins/flyte/skills/flyte-sdk-data/SKILL.md +++ b/plugins/flyte/skills/flyte-sdk-data/SKILL.md @@ -28,6 +28,7 @@ Build ETL pipelines, data processing workflows, and data quality checks with Fly ```python import flyte import flyte.io +import polars as pl env = flyte.TaskEnvironment( name="etl-pipeline", @@ -39,19 +40,20 @@ env = flyte.TaskEnvironment( @env.task(retries=3, cache="auto") async def extract(source_uri: str) -> flyte.io.DataFrame: """Extract data from various sources.""" - import polars as pl if source_uri.endswith(".csv"): df = pl.read_csv(source_uri) elif source_uri.endswith(".parquet"): df = pl.read_parquet(source_uri) else: raise ValueError(f"Unsupported format: {source_uri}") - return flyte.io.DataFrame(df) + # Wrap a native frame rather than calling the raw constructor + return flyte.io.DataFrame.from_df(df) @env.task(retries=2, cache="auto") async def transform(df: flyte.io.DataFrame) -> flyte.io.DataFrame: """Clean and transform data.""" - inner = df.to_polars() + # Materialize a flyte.io.DataFrame as a concrete backend before manipulating it + inner = await df.open(pl.DataFrame).all() cleaned = ( inner .drop_nulls() @@ -60,12 +62,12 @@ async def transform(df: flyte.io.DataFrame) -> flyte.io.DataFrame: pl.col("date").str.strptime(pl.Date, "%Y-%m-%d").alias("date_parsed"), ]) ) - return flyte.io.DataFrame(cleaned) + return flyte.io.DataFrame.from_df(cleaned) @env.task(retries=1, cache="auto") async def load(df: flyte.io.DataFrame, destination: str) -> str: """Load transformed data to destination.""" - inner = df.to_polars() + inner = await df.open(pl.DataFrame).all() if destination.endswith(".parquet"): inner.write_parquet(destination) elif destination.endswith(".csv"): @@ -91,7 +93,7 @@ async def extract_raw(source_uri: str) -> flyte.io.File: df = pl.read_csv(source_uri) path = "/tmp/raw.parquet" df.write_parquet(path) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task(cache="auto") async def validate_raw(raw: flyte.io.File) -> dict: @@ -110,7 +112,7 @@ async def clean(raw: flyte.io.File) -> flyte.io.File: cleaned = df.drop_nulls().unique() path = "/tmp/cleaned.parquet" cleaned.write_parquet(path) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task(cache="auto") async def enrich(cleaned: flyte.io.File) -> flyte.io.File: @@ -120,7 +122,7 @@ async def enrich(cleaned: flyte.io.File) -> flyte.io.File: ... path = "/tmp/enriched.parquet" df.write_parquet(path) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task async def load_enriched(enriched: flyte.io.File, destination: str) -> str: @@ -155,7 +157,7 @@ async def process_file(file_uri: str) -> flyte.io.DataFrame: import polars as pl df = pl.read_parquet(file_uri) cleaned = df.drop_nulls().unique() - return flyte.io.DataFrame(cleaned) + return flyte.io.DataFrame.from_df(cleaned) @env.task async def process_dataset(file_uris: list[str]) -> list: @@ -167,8 +169,8 @@ async def process_dataset(file_uris: list[str]) -> list: async def merge_results(results: list) -> flyte.io.DataFrame: """Merge processed results into a single DataFrame.""" import polars as pl - combined = pl.concat([r.to_polars() for r in results]) - return flyte.io.DataFrame(combined) + combined = pl.concat([await r.open(pl.DataFrame).all() for r in results]) + return flyte.io.DataFrame.from_df(combined) @env.task async def main(file_uris: list[str]) -> flyte.io.DataFrame: @@ -184,7 +186,7 @@ async def process_file_safe(file_uri: str) -> dict: """Process a file with error handling.""" try: df = await process_file(file_uri) - return {"status": "success", "file": file_uri, "rows": len(df.to_polars())} + return {"status": "success", "file": file_uri, "rows": len(await df.open(pl.DataFrame).all())} except Exception as e: return {"status": "error", "file": file_uri, "error": str(e)} @@ -218,10 +220,12 @@ async def main(file_uris: list[str]) -> list: ### Comprehensive data quality ```python +import polars as pl + @env.task(cache="auto") async def validate_schema(df: flyte.io.DataFrame, expected_schema: dict) -> dict: """Validate DataFrame schema matches expected schema.""" - inner = df.to_polars() + inner = await df.open(pl.DataFrame).all() checks = {} # Column names @@ -246,7 +250,7 @@ async def validate_schema(df: flyte.io.DataFrame, expected_schema: dict) -> dict @env.task(cache="auto") async def validate_nulls(df: flyte.io.DataFrame, max_null_pct: float = 0.1) -> dict: """Validate null percentages per column.""" - inner = df.to_polars() + inner = await df.open(pl.DataFrame).all() row_count = len(inner) checks = {} @@ -264,7 +268,7 @@ async def validate_nulls(df: flyte.io.DataFrame, max_null_pct: float = 0.1) -> d @env.task(cache="auto") async def validate_values(df: flyte.io.DataFrame, constraints: dict) -> dict: """Validate value constraints (ranges, enums, patterns).""" - inner = df.to_polars() + inner = await df.open(pl.DataFrame).all() checks = {} for col, constraint in constraints.items(): @@ -310,10 +314,12 @@ async def data_quality_gate( ### Data quality with custom checks ```python +import polars as pl + @env.task(cache="auto") async def check_distribution(df: flyte.io.DataFrame, column: str, expected_stats: dict) -> dict: """Check if data distribution matches expected statistics.""" - inner = df.to_polars() + inner = await df.open(pl.DataFrame).all() col_data = inner[column].drop_nulls() actual_mean = col_data.mean() @@ -356,7 +362,7 @@ async def process_file(file_uri: str) -> flyte.io.File: cleaned = df.drop_nulls() path = f"/tmp/cleaned_{file_uri.split('/')[-1]}" cleaned.write_parquet(path) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task async def merge_files(files: list[flyte.io.File]) -> flyte.io.File: @@ -366,7 +372,7 @@ async def merge_files(files: list[flyte.io.File]) -> flyte.io.File: combined = pl.concat(dfs) path = "/tmp/merged.parquet" combined.write_parquet(path) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task async def dynamic_etl(prefix: str, destination: str) -> dict: @@ -413,35 +419,28 @@ async def conditional_pipeline(df: flyte.io.DataFrame, threshold: float) -> dict ## JsonlFile and JsonlDir for Large Datasets -### JsonlFile — streaming JSONL +The `JsonlFile` and `JsonlDir` types stream records one at a time, so you can process line-delimited data that does not fit in memory. They come from the **`flyteplugins-jsonl`** plugin (which must be in the image) and extend the built-in `flyte.io.File` / `flyte.io.Dir`, inheriting remote storage and caching. Every method has a `_sync` counterpart. ```python +# Streaming writer to remote object storage +@env.task +async def write_records() -> None: + from flyteplugins.jsonl import JsonlFile + out = JsonlFile.new_remote("results.jsonl") + async with out.writer() as w: + await w.write({"id": 1}) # also write_many(), write_raw(), flush() + +# Streaming read of a sharded JSONL directory @env.task -async def process_jsonl(path: str) -> int: - """Process a JSONL file with streaming.""" - from flyte.extend import JsonlFile - jf = JsonlFile(path) +async def read_jsonl_dir(d: flyte.io.Dir) -> int: + from flyteplugins.jsonl import JsonlDir + jd = JsonlDir(d) count = 0 - async for record in jf.stream(): - process(record) + async for rec in jd.iter_records(): count += 1 return count ``` -### JsonlDir — batched JSONL directories - -```python -@env.task -async def process_jsonl_dir(dir_path: str) -> dict: - """Process JSONL directory with batched streaming.""" - from flyte.extend import JsonlDir - jd = JsonlDir(dir_path) - total = 0 - async for batch in jd.stream_batches(): - total += len(batch) - return {"records": total} -``` - ## Data Format Reference | Format | Flyte Type | Best For | @@ -469,5 +468,6 @@ async def process_jsonl_dir(dir_path: str) -> dict: 1. **Don't load entire datasets into memory** — use streaming (`JsonlFile`, Polars lazy) for large data. 2. **Don't pass DataFrames inline** — they go by reference automatically, but small dicts do go inline. -3. **Don't skip data quality gates** — always validate before and after transforms. -4. **Don't use Union-only features** — avoid `ReusePolicy` and other Union-specific APIs. +3. **Don't use `.to_polars()` / raw `DataFrame(...)` constructors** — wrap with `flyte.io.DataFrame.from_df(...)` and materialize with `await df.open(pl.DataFrame).all()`; create files with `File.from_local(...)` / `File.new_remote()` (the raw `File(path=...)` constructor is only for existing remote refs). +4. **Don't skip data quality gates** — always validate before and after transforms. +5. **Don't use Union-only features** — avoid `ReusePolicy` and other Union-specific APIs. diff --git a/plugins/flyte/skills/flyte-sdk-eval/SKILL.md b/plugins/flyte/skills/flyte-sdk-eval/SKILL.md index e19c636..01e43a0 100644 --- a/plugins/flyte/skills/flyte-sdk-eval/SKILL.md +++ b/plugins/flyte/skills/flyte-sdk-eval/SKILL.md @@ -113,7 +113,7 @@ env = flyte.TaskEnvironment( async def load_test_data() -> flyte.io.DataFrame: """Load ground truth test data.""" ... - return flyte.io.DataFrame(df) + return flyte.io.DataFrame.from_df(df) @env.task async def load_model(model_uri: str) -> object: @@ -135,9 +135,10 @@ async def compute_metrics( accuracy_score, f1_score, precision_score, recall_score, roc_auc_score, mean_squared_error, ) - y_true = ground_truth.to_polars()["label"].to_list() - y_pred = predictions.to_polars()["prediction"].to_list() - y_prob = predictions.to_polars()["probability"].to_list() + import polars as pl + y_true = (await ground_truth.open(pl.DataFrame).all())["label"].to_list() + y_pred = (await predictions.open(pl.DataFrame).all())["prediction"].to_list() + y_prob = (await predictions.open(pl.DataFrame).all())["probability"].to_list() return { "accuracy": accuracy_score(y_true, y_pred), @@ -155,7 +156,7 @@ async def generate_report(metrics: dict) -> flyte.io.File: # Confusion matrix, ROC curve, etc. path = "/tmp/eval_report.png" fig.savefig(path) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task async def evaluate_pipeline( @@ -203,7 +204,8 @@ async def compare_models( @env.task async def validate_data(df: flyte.io.DataFrame) -> dict: """Run data quality checks.""" - inner = df.to_polars() + import polars as pl + inner = await df.open(pl.DataFrame).all() checks = {} # Row count check @@ -295,7 +297,7 @@ async def track_experiment( path = f"/tmp/experiments/{experiment_name}.json" with open(path, "w") as f: json.dump(record, f, indent=2) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task async def run_experiment( @@ -424,7 +426,7 @@ tests/ ## Anti-Patterns 1. **Don't test against remote runs in unit tests** — use direct function invocation for unit tests. Reserve `flyte.run()` for integration tests. -2. **Don't hardcode test data paths** — use `flyte.io.File(path="/tmp/test_data")` with temp directories. +2. **Don't hardcode test data paths** — prefer `File.from_local(...)` / `File.from_existing_remote(...)` with temp directories over bare path strings. 3. **Don't skip data quality gates** — always validate data before and after transformations. 4. **Don't use Union-only features** — avoid `ReusePolicy` and other Union-specific APIs. 5. **Don't test ML models with random data** — use representative test datasets that match production distribution. diff --git a/plugins/flyte/skills/flyte-sdk-ml/SKILL.md b/plugins/flyte/skills/flyte-sdk-ml/SKILL.md index e860d00..3b75639 100644 --- a/plugins/flyte/skills/flyte-sdk-ml/SKILL.md +++ b/plugins/flyte/skills/flyte-sdk-ml/SKILL.md @@ -65,7 +65,7 @@ async def train( output_path = "/tmp/model_checkpoint" model.save_pretrained(output_path) tokenizer.save_pretrained(output_path) - return flyte.io.File(path=output_path) + return await flyte.io.File.from_local(output_path) @env.task async def main( @@ -77,8 +77,8 @@ async def main( ) -> dict: hyperparams = {"lr": lr, "batch_size": batch_size, "epochs": epochs} checkpoint = await train( - train_data=flyte.io.File(path=train_uri), - val_data=flyte.io.File(path=val_uri), + train_data=flyte.io.File.from_existing_remote(train_uri), + val_data=flyte.io.File.from_existing_remote(val_uri), hyperparams=hyperparams, ) return {"checkpoint": checkpoint, "hyperparams": hyperparams} @@ -125,7 +125,7 @@ async def train_sklearn( path = f"/tmp/{model_type}_model.joblib" joblib.dump(model, path) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) ``` ### HuggingFace Trainer @@ -194,7 +194,7 @@ async def train_hf( trainer.save_model("/tmp/final_model") tokenizer.save_pretrained("/tmp/final_model") - return flyte.io.File(path="/tmp/final_model") + return await flyte.io.File.from_local("/tmp/final_model") ``` ## Hyperparameter Optimization @@ -303,7 +303,7 @@ async def track_experiment( path = f"/tmp/experiments/{experiment_name}_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(path, "w") as f: json.dump(record, f, indent=2) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task async def compare_experiments( @@ -339,7 +339,7 @@ async def track_inference( path = f"/tmp/inference/{model_uri.split('/')[-1]}_{datetime.datetime.now().strftime('%Y%m%d')}.json" with open(path, "w") as f: json.dump(record, f, indent=2) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) ``` ## Model Evaluation and Selection @@ -430,7 +430,7 @@ async def generate_comparison_report( path = "/tmp/model_comparison.png" fig.savefig(path, bbox_inches="tight") - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) ``` ## Batch Inference @@ -496,7 +496,7 @@ async def batch_predict( results = pl.DataFrame({"prediction": all_preds, "probability": all_probs}) path = f"/tmp/predictions_{data_file.path.split('/')[-1]}" results.write_parquet(path) - return flyte.io.File(path=path) + return await flyte.io.File.from_local(path) @env.task async def batch_inference( @@ -507,7 +507,7 @@ async def batch_inference( model_ctx = await load_model(model_uri) # Fan out inference across files results = await flyte.map( - lambda f: batch_predict(model_ctx, flyte.io.File(path=f)), + lambda f: batch_predict(model_ctx, flyte.io.File.from_existing_remote(f)), data_files, ) return results diff --git a/plugins/flyte/skills/flyte-sdk-run/SKILL.md b/plugins/flyte/skills/flyte-sdk-run/SKILL.md index 31cf8d6..b3ffe60 100644 --- a/plugins/flyte/skills/flyte-sdk-run/SKILL.md +++ b/plugins/flyte/skills/flyte-sdk-run/SKILL.md @@ -200,31 +200,117 @@ result = flyte.run(main, inputs={"data": ["a"]}) print(result.url) # open the run in the UI to view logs ``` -## Re-running Runs +## Re-running Runs (rerun) + +Every run is durable: its task, code, inputs, and configuration are recorded on the backend. `flyte rerun` / `flyte.rerun()` relaunches a new run from a previous one **without needing the original code checked out locally** — task and inputs are fetched from the platform. ### CLI ```bash -# Re-run with original code and inputs -flyte rerun --project flytesnacks --domain development +# Rerun with the prior run's exact code and inputs +flyte rerun + +# Give the new run a name and stream its parent-action logs +flyte rerun --name retry-1 --follow + +# Override an env var / set a label on the new run +flyte rerun -e LOG_LEVEL=20 --label team=ml -# Re-run with new local code -flyte run --rerun-from pipeline.py main --data '[4,5,6]' +# Rerun with a different input value (converted against the prior run's interface) +flyte rerun --threshold 0.9 ``` +| Option | Description | +|---|---| +| `-p` / `-d` | Project / domain for the new run | +| `--name` | Name for the new run | +| `-e KEY=VALUE` | Override an environment variable (repeatable) | +| `--label KEY=VALUE` | Set a label (repeatable) | +| `-f` / `--follow` | Stream the parent action's logs after launch | + +> [!NOTE] `flyte rerun` is **not** `flyte run`. `flyte run ` uses your local code and CLI inputs; `flyte rerun ` fetches both from the backend. To keep the successful actions of a failed run and re-execute only what failed, add `--recover` — see **Recover a failed run** below. + ### Python SDK ```python import flyte +flyte.init_from_config() -# Re-run with new inputs -result = flyte.run( - main, - inputs={"data": [4, 5, 6]}, - run_context=flyte.with_runcontext(run_name="rerun-of-abc123"), -) +# Rerun with the prior run's exact inputs (task + inputs fetched from the platform) +r = flyte.rerun("ul56wcvgqrb9vzhzz5l2") + +# Change inputs — converted against the prior run's interface +r = flyte.rerun("ul56wcvgqrb9vzhzz5l2", x_list=[1, 2, 3]) + +# Control launch settings with flyte.with_runcontext(...) +flyte.with_runcontext(name="retry-1", env_vars={"LOG_LEVEL": "20"}).rerun("ul56wcvgqrb9vzhzz5l2") ``` +`flyte.rerun()` returns a `flyte.remote.Run`, so you can wait on it and read outputs exactly like `flyte.run()`. + +## Recover a failed run + +**Recovery** launches a new run that **reuses the successful actions** of a previous run and re-executes only what failed or never ran. A reused action is not executed again: its recorded output is handed to whatever consumes it. Recovery is remote-only (not supported in `--local`). + +Recovery differs from caching: + +- **Caching** is content-addressed (keyed on inputs + task version), enabled per task in code, reusable across any run. +- **Recovery** is scoped to one named run, needs **no** task preparation in code, and is opted in per run at launch (`--recover`). + +### Recover from the CLI + +```bash +# Reuse the prior run's succeeded actions, re-running only what failed or never ran +flyte rerun --recover --follow + +# Force specific actions to re-execute even though they succeeded +flyte rerun --recover --force-rerun-action a3 --force-rerun-action a7 + +# Re-run a single action on its own, rooted at its task with its recorded inputs +flyte rerun --action-name a3 + +# Proceed even if the source run's outputs were cleaned up from storage +flyte rerun --recover --allow-missing-outputs +``` + +| Option | Description | +|---|---| +| `--recover` | Reuse succeeded actions; re-run only what failed or never ran | +| `--action-name ` | Re-run only this action (cannot be combined with `--recover`) | +| `--force-rerun-action ` | With `--recover`, re-execute an action that already succeeded (repeatable) | +| `--allow-missing-outputs` | Proceed when the source run's outputs were cleaned up | +| `-e` / `--env KEY=VALUE` | Set an env var on the new run | + +Action names are deterministic hashes, not positions — list them with `flyte get action `. Reused actions land in the `RECOVERED` phase (terminal, success-equivalent), so `flyte get action ` shows exactly what recovery skipped. + +> [!WARNING] Recovered actions keep their **original** outputs. Recovery does not recompute them against new inputs — name the actions that must re-execute in `--force-rerun-action`. + +### Recover programmatically + +```python +import flyte +flyte.init_from_config() + +# Re-run everything with the prior run's inputs +flyte.rerun("") + +# Reuse the succeeded actions, re-running only what failed or never ran +flyte.rerun("", recover=True) + +# Force specific actions to re-execute even though they succeeded +flyte.rerun("", recover=True, force_rerun_actions=["a3", "a7"]) + +# Change inputs, keeping the ones you leave out, and still reuse succeeded actions +flyte.rerun("", recover=True, threshold=0.9) + +# Re-run a single action rooted at its task +flyte.rerun("", action_name="a3") + +flyte.with_runcontext(name="retry-1", env_vars={"LOG_LEVEL": "20"}).rerun("", recover=True) +``` + +> [!NOTE] Task inputs share the keyword namespace with these arguments, so a task input named `run_name`, `action_name`, `recover`, `force_rerun_actions`, or `allow_missing_source_outputs` cannot be passed via `flyte.rerun`. + ## Running Tasks (vs Workflows) ### Run a single task @@ -346,7 +432,7 @@ async def get_run_data(run_name: str) -> flyte.io.File: @env.task async def upload_local_data(file_path: str) -> flyte.io.File: """Upload local file to remote storage for a run.""" - return flyte.io.File(path=file_path) + return await flyte.io.File.from_local(file_path) ``` ### S3 / GCS / Azure access diff --git a/plugins/flyte/skills/flyte-sdk-types/SKILL.md b/plugins/flyte/skills/flyte-sdk-types/SKILL.md index 87a439f..cc15d35 100644 --- a/plugins/flyte/skills/flyte-sdk-types/SKILL.md +++ b/plugins/flyte/skills/flyte-sdk-types/SKILL.md @@ -38,135 +38,197 @@ Flyte 2 uses **Python type hints** for serialization. Every task input/output mu | `pydantic.BaseModel` | Struct | JSON in metadata bucket | | `datetime`, `timedelta` | DateTime / Duration | Inline | -## flyte.io.File — Single Files +## Offloaded types: flyte.io.File and flyte.io.Dir -Use `flyte.io.File` for any single file that flows between tasks. Files are **automatically uploaded** to the metadata bucket at runtime. +`flyte.io.File` and `flyte.io.Dir` are the **offloaded types**. A variable of an offloaded type holds a *reference* to data, not the data itself. The bytes live in the internal blob store; they are uploaded when first created and only downloaded when a task actually accesses them. This lets very large objects (videos, DNA datasets, model weights) move between tasks without transferring bytes. + +Use `flyte.io.File` for a single file and `flyte.io.Dir` for a directory of files. ```python +import tempfile +from pathlib import Path import flyte -import flyte.io +from flyte.io import Dir, File + +env = flyte.TaskEnvironment(name="files-and-dirs") @env.task -async def download_url(url: str) -> flyte.io.File: - """Download a file and return as flyte.io.File.""" - import urllib.request - local_path = f"/tmp/{url.split('/')[-1]}" - urllib.request.urlretrieve(url, local_path) - return flyte.io.File(path=local_path) +async def write_file(name: str) -> File: + with open("test.txt", "w") as fh: + fh.write(f"hello world {name}") + # Upload the local file to the blob store and return the reference. + return await File.from_local("test.txt") @env.task -async def process(file: flyte.io.File) -> dict: - """Read a file — flyte.io.File downloads it automatically.""" - # file.path gives the local path (already downloaded) - with open(file.path) as f: - content = f.read() - return {"lines": len(content.splitlines())} +async def check_dir(my_dir: Dir): + # Stream each file out of the directory without downloading it whole. + async for file in my_dir.walk(): + async with file.open("rb") as fh: + print(file.name, await fh.read()) @env.task -async def main(url: str) -> dict: - downloaded = await download_url(url) - return await process(downloaded) # type: flyte.io.File flows as remote reference +async def main(): + f = await write_file("Alice") + temp_dir = tempfile.mkdtemp() + async with f.open("rb") as fh: + (Path(temp_dir) / f.name).write_bytes(await fh.read()) + my_dir = await Dir.from_local(temp_dir) + await check_dir(my_dir=my_dir) ``` -### flyte.io.File best practices +### Synchronous and asynchronous APIs + +Every I/O operation on `File`/`Dir` has an async form (used in `async def` tasks) and a `_sync` form (used in plain `def` tasks). Pick the one matching how your task is defined: + +- **Async task** (`async def`): `await File.from_local(...)`, `async with file.open(...)`, `async for f in dir.walk()`, `await file.download()`, `await file.exists()`. +- **Sync task** (`def`): `File.from_local_sync(...)`, `with file.open_sync(...)`, `for f in dir.walk_sync()`, `file.download_sync()`, `file.exists_sync()`. + +Constructors like `File.new_remote()` and `File.from_existing_remote(uri)` do no I/O, so a single form works from either kind of task. + +#### File methods + +| Async (in `async def`) | Sync (in `def`) | Purpose | +|---|---|---| +| `await File.from_local(path)` | `File.from_local_sync(path)` | Upload a local file to the blob store | +| `File.new_remote()` | `File.new_remote()` | Allocate a new remote file to stream into | +| `File.from_existing_remote(uri)` | `File.from_existing_remote(uri)` | Reference a file already stored remotely | +| `async with file.open(mode) as fh` | `with file.open_sync(mode) as fh` | Stream the file for reading/writing | +| `await file.download()` | `file.download_sync()` | Download to the local filesystem | +| `await file.exists()` | `file.exists_sync()` | Check whether the file exists | + +#### Dir methods -- **Always pass `flyte.io.File` between tasks** — never pass file paths as strings. Flyte serializes the reference to the remote blob. -- **`file.path`** gives the local download path inside the task container. -- **Don't hardcode paths** — let Flyte manage the download/upload lifecycle. -- **Compression**: Flyte infers format from extension (`.parquet`, `.csv`, `.json`, `.pt`, `.png`, etc.). +| Async (in `async def`) | Sync (in `def`) | Purpose | +|---|---|---| +| `await Dir.from_local(path)` | `Dir.from_local_sync(path)` | Upload a local directory | +| `Dir.new_remote()` | `Dir.new_remote()` | Allocate a new remote directory | +| `Dir.from_existing_remote(uri)` | `Dir.from_existing_remote(uri)` | Reference a directory stored remotely | +| `async for f in dir.walk()` | `for f in dir.walk_sync()` | Iterate over the files (supports `file_pattern`, `recursive`, `max_depth`) | +| `await dir.list_files()` | `dir.list_files_sync()` | List files (non-recursive) | +| `await dir.get_file(name)` | `dir.get_file_sync(name)` | Get one file by name | +| `await dir.download()` | `dir.download_sync()` | Download the whole directory | +| `await dir.exists()` | `dir.exists_sync()` | Check whether the directory exists | -## flyte.io.Dir — Directories +> [!NOTE] `walk()` yields `File` objects streaming from the directory; it does not require downloading the whole directory. If a `File`/`Dir` already points at the local filesystem, `download()`/`download_sync()` return its own path without copying (Flyte 2.11.2+); pass a different `local_path` to force a copy. -Use `flyte.io.Dir` for a collection of files (e.g., model checkpoints, output artifacts). +### Synchronous example ```python import flyte -import flyte.io +from flyte.io import File + +env = flyte.TaskEnvironment(name="sync-file") @env.task -async def train(checkpoint_dir: flyte.io.Dir) -> flyte.io.Dir: - """Train and save checkpoints to a directory.""" - # Write checkpoints - for i in range(10): - path = f"{checkpoint_dir.path}/checkpoint_{i}.pt" - save_model(path) - return checkpoint_dir +def write_file(content: str) -> File: + f = File.new_remote() # no I/O, single form + with f.open_sync("wb") as fh: # stream into it, blocking call + fh.write(content.encode("utf-8")) + return f @env.task -async def evaluate(checkpoints: flyte.io.Dir) -> dict: - """Load checkpoints from a directory.""" - # List all files in the directory - files = list(checkpoints.path.glob("*.pt")) - ... +def read_file(f: File) -> str: + with f.open_sync("rb") as fh: + return fh.read().decode("utf-8") + +@env.task +def main() -> str: + return read_file(write_file("hello world")) ``` -## flyte.io.DataFrame — Polars DataFrames +### flyte.io.File / Dir best practices + +- **Always pass `flyte.io.File`/`flyte.io.Dir` between tasks** — never pass file paths as strings. Flyte serializes the reference to the remote blob and downloads it on access. +- Upload a local file with `await File.from_local(path)` (async) or `File.from_local_sync(path)` (sync); read/write by streaming with `open()`/`open_sync()`. +- **`file.path`** gives the local path inside the task container (after download/access); `file.name` is the basename. +- **Content-based caching**: because offloaded types are passed by reference, a cached downstream task does **not** hit on identical content at a new path. To cache on content, attach a hash — see [Content-based caching](#content-based-caching-for-dataframes-files-and-directories) below. +- Compress format is inferred from the extension (`.parquet`, `.csv`, `.json`, `.bin`, etc.). + +## flyte.io.DataFrame — DataFrames (Pandas, Polars, Arrow, Dask) -Flyte 2 has built-in support for Polars DataFrames. They are **passed by reference** (Parquet in the metadata bucket), not inline. +Flyte auto-serializes these native DataFrame types between tasks (stored as Parquet in the blob store, passed **by reference**): `pandas.DataFrame`, `pyarrow.Table`, `dask.dataframe.DataFrame`, `polars.DataFrame`, and `flyte.io.DataFrame`. No annotation is required for a native type — returning one uploads it at task completion. + +> [!NOTE] **Polars needs a plugin.** Native `pandas`/`pyarrow`/`dask` work out of the box. `polars.DataFrame` and `polars.LazyFrame` (and `flyte.io.DataFrame` of a Polars frame) require the **`flyteplugins-polars`** package in the image; it registers itself on install. Lance datasets require `flyteplugins-lance`. Add them with `flyte.Image.from_debian_base().with_pip_packages("flyteplugins-polars", ...)`. ```python +import pandas as pd import flyte import flyte.io @env.task -async def load_csv(url: str) -> flyte.io.DataFrame: - """Load a CSV and return as Polars DataFrame.""" - import polars as pl - df = pl.read_csv(url) - return flyte.io.DataFrame(df) +async def produce() -> pd.DataFrame: + return pd.DataFrame({"id": [1, 2, 3], "value": [10, 20, 30]}) @env.task -async def clean(df: flyte.io.DataFrame) -> flyte.io.DataFrame: - """Clean the DataFrame.""" - inner = df.to_polars() # Get the underlying Polars DataFrame - cleaned = inner.drop_nulls() - return flyte.io.DataFrame(cleaned) +async def consume(df: pd.DataFrame) -> int: + # A native pandas.DataFrame input is downloaded and materialized for you. + return int(df["value"].sum()) +``` + +### flyte.io.DataFrame wrapper + +`flyte.io.DataFrame` is a thin, uniform wrapper around any standard DataFrame type, useful when a task must accept or emit several engine types or you want to pin the storage format. Create it with `from_df` and pin the format with an `Annotated` type: + +```python +from typing import Annotated +import pandas as pd +import flyte +import flyte.io @env.task -async def save_parquet(df: flyte.io.DataFrame, path: str) -> flyte.io.File: - """Save DataFrame to Parquet.""" - inner = df.to_polars() - inner.write_parquet(path) - return flyte.io.File(path=path) +async def create() -> Annotated[flyte.io.DataFrame, "parquet"]: + return flyte.io.DataFrame.from_df(pd.DataFrame({"a": [1, 2, 3]})) @env.task -async def main(url: str) -> flyte.io.DataFrame: - raw = await load_csv(url) - cleaned = await clean(raw) - return cleaned # Flows as Parquet reference +async def read(df: flyte.io.DataFrame) -> float: + # Materialize as a specific backend with open(...).all() + pdf = await df.open(pd.DataFrame).all() + return float(pdf["a"].mean()) ``` -### Polars DataFrame patterns +Flyte converts a `flyte.io.DataFrame` into whatever concrete type a task declares as its input (for example `pd.DataFrame`) on access. To get a concrete backend from a `flyte.io.DataFrame` inside a task, `await df.open(pd.DataFrame).all()`. -```python -# Convert to Polars for manipulation -inner_df = df.to_polars() +### Eager vs Lazy workflows -# Convert from Polars -df = flyte.io.DataFrame(inner_df) +- **Pandas / eager `pl.DataFrame`**: materialized on use. +- **Lazy (`pl.LazyFrame` / `dask`)**: the query plan is deferred until execution; Flyte serializes the result to Parquet when it is passed on. Call `collect()` in the producing task to trigger execution if needed. -# Common operations -df = flyte.io.DataFrame(inner_df.filter(pl.col("age") > 18)) -df = flyte.io.DataFrame(inner_df.group_by("category").agg(pl.col("value").mean())) - -# Check shape -print(df.shape) # (rows, cols) -print(df.schema) # column names and types -``` +## Polars DataFrames (plugin) -### Eager vs Lazy DataFrames +`polars.DataFrame` and `polars.LazyFrame` support comes from the **`flyteplugins-polars`** package, which must be installed in the task image. Both types are serialized as Parquet between tasks, just like the other DataFrame backends. ```python -# Eager (loaded into memory) -df = flyte.io.DataFrame(inner_df) +import polars as pl +import flyte + +env = flyte.TaskEnvironment( + name="polars-dataframes", + image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-polars", "polars"), + resources=flyte.Resources(cpu="1", memory="2Gi"), +) + +@env.task +async def create_dataframe() -> pl.DataFrame: + """Polars DataFrames pass between tasks as Parquet — no manual upload.""" + return pl.DataFrame({"dept": ["eng", "fin"], "salary": [75_000, 68_000]}) + +@env.task +async def filter_high_earners(df: pl.DataFrame) -> pl.DataFrame: + return df.filter(pl.col("salary") > 60_000).sort("salary", descending=True) -# Lazy (streaming, for large datasets) -df = flyte.io.DataFrame(inner_df.lazy()) +@env.task +async def create_lazyframe() -> pl.LazyFrame: + """LazyFrames defer the plan; Flyte serializes the result to Parquet.""" + return pl.LazyFrame({"dept": ["eng", "fin"], "salary": [75_000, 68_000]}) -# Materialize lazy to eager -eager = df.to_polars() # materializes +@env.task +async def aggregate(lf: pl.LazyFrame) -> pl.DataFrame: + # collect() executes the lazy plan + return lf.group_by("dept").agg(pl.col("salary").mean()).collect() ``` +> [!NOTE] Flyte's `flyte.io.DataFrame` converter does **not** auto-load in a task pinned to `pl.LazyFrame` unless the polars plugin is present; install `flyteplugins-polars` first. + ## Dataclass and Pydantic Models For structured data, use Python dataclasses or Pydantic models. Flyte serializes them to JSON. @@ -204,59 +266,61 @@ async def predict(model: flyte.io.File, data: flyte.io.DataFrame) -> PredictionO ## Custom Type Transformers -Extend Flyte's type system to support custom types (e.g., PIL Images, HuggingFace datasets). +Flyte's type system handles most Python types automatically. If it does not recognize a type it falls back to **pickle**, which is inefficient, Python-only, version-fragile, and opaque in the UI. For data that deserves a real format (Parquet, tensor formats, etc.), register a type transformer. -### PIL Image transformer +There are two kinds of extensions: -```python -from PIL import Image -import flyte -import flyte.io -from flyte.types import TypeTransformer - -class PILImageTransformer(TypeTransformer[Image.Image]): - _type = Image.Image - - def get_type(self, input: Image.Image) -> type: - return Image.Image - - def save(self, img: Image.Image, path: str) -> None: - img.save(path) - - def load(self, path: str) -> Image.Image: - return Image.open(path) - -# Register the transformer -flyte.types.TypeEngine.register(PILImageTransformer()) - -# Now use it in tasks -@env.task -async def process_image(img: Image.Image) -> flyte.io.File: - # img is a PIL Image, already downloaded - ... -``` +1. **Type transformers** — for scalar types (custom objects, files, directories). +2. **DataFrame extensions** — for tabular types with schema metadata (Parquet, parallel/partitioned writes). -### HuggingFace Dataset transformer +A Flyte 2 type transformer subclasses `TypeTransformer[T]` and implements `get_literal_type`, `to_literal`, `to_python_value`, and `guess_python_type`. The method signatures use Flyte literal protobuf types and are `async`: ```python -from datasets import Dataset -import flyte - -class HFDatasetTransformer(TypeTransformer[Dataset]): - _type = Dataset - - def get_type(self, input: Dataset) -> type: - return Dataset - - def save(self, ds: Dataset, path: str) -> None: - ds.save_to_disk(path) - - def load(self, path: str) -> Dataset: - return Dataset.load_from_disk(path) - -flyte.types.TypeEngine.register(HFDatasetTransformer()) +from flyteidl2.core import literals_pb2, types_pb2 +from flyte.types import TypeEngine, TypeTransformer, TypeTransformerFailedError + +class PositiveInt: + """Example custom type: an int that must be positive.""" + def __init__(self, value: int): + if value <= 0: + raise ValueError(f"expected positive, got {value}") + self._value = value + @property + def value(self) -> int: + return self._value + +class PositiveIntTransformer(TypeTransformer[PositiveInt]): + def __init__(self): + super().__init__(name="PositiveInt", t=PositiveInt) + + def get_literal_type(self, t): + return types_pb2.LiteralType( + simple=types_pb2.SimpleType.INTEGER, + structure=types_pb2.TypeStructure(tag="PositiveInt"), + ) + + async def to_literal(self, python_val, python_type, expected): + if not isinstance(python_val, PositiveInt): + raise TypeTransformerFailedError(...) + return literals_pb2.Literal( + scalar=literals_pb2.Scalar(primitive=literals_pb2.Primitive(integer=python_val.value)) + ) + + async def to_python_value(self, lv, expected_python_type): + return PositiveInt(lv.scalar.primitive.integer) + + def guess_python_type(self, literal_type): + if literal_type.simple == types_pb2.SimpleType.INTEGER \ + and literal_type.structure.tag == "PositiveInt": + return PositiveInt + raise ValueError(f"cannot guess PositiveInt from {literal_type}") + +# Register it (call before task execution) +TypeEngine.register(PositiveIntTransformer()) ``` +> [!WARNING] The Flyte 1 style `save()`/`load()`/`get_type()` `TypeTransformer` shown in older examples is **obsolete**. Flyte 2 uses the `get_literal_type` / `async to_literal` / `async to_python_value` / `guess_python_type` protocol above. + ## Data I/O Patterns by Domain ### ETL / Data Engineering @@ -268,28 +332,25 @@ async def csv_to_parquet(csv_file: flyte.io.File, output_path: str) -> flyte.io. import polars as pl df = pl.read_csv(csv_file.path) df.write_parquet(output_path) - return flyte.io.File(path=output_path) + return await flyte.io.File.from_local(output_path) -# JsonlFile — batched JSONL reading +# JsonlFile — streaming JSONL (needs the flyteplugins-jsonl plugin in your image) @env.task -async def process_jsonl(path: str) -> int: - from flyte.extend import JsonlFile - jf = JsonlFile(path) - count = 0 - async for record in jf.stream(): - process(record) - count += 1 - return count - -# JsonlDir — batched JSONL directory +async def write_records() -> None: + from flyteplugins.jsonl import JsonlFile + out = JsonlFile.new_remote("results.jsonl") # streaming writer to remote storage + async with out.writer() as w: + await w.write({"id": 1}) # also write_many(), write_raw(), flush() + +# JsonlDir — directory of sharded JSONL @env.task -async def process_jsonl_dir(dir_path: str) -> dict: - from flyte.extend import JsonlDir - jd = JsonlDir(dir_path) +async def read_jsonl_dir(d: flyte.io.Dir) -> int: + from flyteplugins.jsonl import JsonlDir + jd = JsonlDir(d) total = 0 - async for batch in jd.stream_batches(): - total += len(batch) - return {"records": total} + async for rec in jd.iter_records(): + total += 1 + return total ``` ### Image Processing @@ -305,7 +366,7 @@ async def resize_image(input_file: flyte.io.File, size: tuple[int, int]) -> flyt resized = img.resize(size) output_path = f"/tmp/resized_{size[0]}x{size[1]}.png" resized.save(output_path) - return flyte.io.File(path=output_path) + return await flyte.io.File.from_local(output_path) @env.task async def batch_resize(files: list[flyte.io.File], size: tuple[int, int]) -> list: @@ -329,7 +390,7 @@ async def extract_features(audio_file: flyte.io.File) -> flyte.io.File: # Save as numpy array output_path = "/tmp/mfccs.npz" np.savez(output_path, mfccs=mfccs, sr=sr) - return flyte.io.File(path=output_path) + return await flyte.io.File.from_local(output_path) ``` ### HuggingFace Datasets @@ -339,22 +400,60 @@ from datasets import load_dataset, Dataset import flyte import flyte.io -@env_task -async def load_dataset_from_hub(dataset_name: str, split: str = "train") -> flyte.io.File: - """Load a HF dataset and save locally.""" +@env.task +async def load_dataset_from_hub(dataset_name: str, split: str = "train") -> flyte.io.Dir: + """Load a HF dataset and save locally, then upload as a Dir.""" ds = load_dataset(dataset_name, split=split) path = f"/tmp/{dataset_name.replace('/', '_')}_{split}" ds.save_to_disk(path) - return flyte.io.File(path=path) + return await flyte.io.Dir.from_local(path) @env.task -async def process_dataset(file: flyte.io.File) -> flyte.io.DataFrame: - """Convert HF dataset to Flyte DataFrame.""" - ds = Dataset.load_from_disk(file.path) - df = ds.to_pandas() - return flyte.io.DataFrame(df) +async def process_dataset(d: flyte.io.Dir) -> flyte.io.DataFrame: + """Convert HF dataset to a Flyte DataFrame.""" + ds = Dataset.load_from_disk(d.path) + return flyte.io.DataFrame.from_df(ds.to_pandas()) +``` + +## Distributing a type transformer as a plugin + +To share a transformer as an installable package, register it through a `pyproject.toml` entry point so Flyte auto-loads it on `flyte.init()` / `flyte.init_from_config()`: + +```toml +[project.entry-points."flyte.plugins.types"] +my_transformer = "my_transformer.transformer:register_positive_int_transformer" ``` +The entry point group `flyte.plugins.types` tells the SDK to load the transformer when the package is installed. To keep init fast you can disable auto-loading with `flyte.init(load_plugin_type_transformers=False)` (default `True`). + +## Content-based caching for DataFrames, files, and directories + +Offloaded types are keyed by **remote path**, so a downstream cached task does *not* hit on identical content stored at a new path. To cache on content, attach a hash at production time with `flyte.io.HashFunction`: + +```python +from typing import Annotated +import hashlib +import pandas as pd +import flyte +import flyte.io +from flyte.io import HashFunction + +# Native DataFrame: annotate the return type with a HashFunction +HashedPandas = Annotated[pd.DataFrame, HashFunction.from_fn(lambda df: str(hash(tuple(df["value"]))))] + +@env.task +async def produce() -> HashedPandas: + return pd.DataFrame({"value": [1, 2, 3]}) + +@env.task(cache="auto") +async def consume(df: pd.DataFrame) -> int: # cache key is the content hash + return int(df["value"].sum()) +``` + +- **`flyte.io.DataFrame`**: pass the `HashFunction` to `DataFrame.from_local(df, hash_method=...)` instead of annotating. +- **`flyte.io.File`**: `File.from_local(path, hash_method=HashFunction.from_fn(hash_bytes))` (also on `File.new_remote`). +- **`flyte.io.Dir`**: `Dir.from_local` does *not* take a `HashFunction`; compute a content key yourself and pass it as `dir_cache_key=`. + ## Data Locality and Storage Best Practices ### How data flows between tasks @@ -370,7 +469,7 @@ async def process_dataset(file: flyte.io.File) -> flyte.io.DataFrame: | `list`, `dict` (small) | Inline | ~10 MB | | `flyte.io.File` | Reference (S3/GCS) | Unlimited | | `flyte.io.Dir` | Reference (S3/GCS) | Unlimited | -| `flyte.io.DataFrame` (Polars) | Reference (Parquet) | Unlimited | +| `flyte.io.DataFrame` (Pandas/Polars/Arrow/Dask) | Reference (Parquet) | Unlimited | | `dataclass` / `BaseModel` | Inline (JSON) | ~10 MB | ### Storage best practices @@ -384,7 +483,7 @@ async def process_dataset(file: flyte.io.File) -> flyte.io.DataFrame: ## Inline I/O Threshold -Control when data is passed inline vs by reference: +Control when data is passed inline vs by reference. Offloaded types (`File`, `Dir`, `DataFrame`) always go by reference; the threshold governs when large materialized values (big `list`/`dict`/dataclass) switch to a blob instead of inline JSON: ```python @env.task(inline_output_limit="10MB") # data > 10MB goes by reference @@ -397,7 +496,8 @@ Default threshold is generous. For ML training outputs or large DataFrames, set ## Common Type Mistakes 1. **Missing type hints** — Flyte 2 requires type annotations on all task inputs/outputs. No type hint = serialization error. -2. **Passing `flyte.io.File` as a string** — always use `flyte.io.File(path=...)` objects. The path string alone won't serialize. -3. **Using Pandas instead of Polars** — Flyte 2's native DataFrame is Polars. Use `df.to_polars()` to get the underlying DataFrame. -4. **Not registering custom transformers** — if you register a custom type transformer, it must be registered before task execution. -5. **Forgetting `.path` on flyte.io.File** — inside a task, `file` is a `flyte.io.File` object, not a string. Use `file.path` for the local path. +2. **Passing a path string where an offloaded type is expected** — use `flyte.io.File`/`flyte.io.Dir` objects (create with `File.from_local(...)` / `Dir.from_local(...)`). The path string alone won't serialize. +3. **Forgetting a DataFrame plugin** — `pandas`/`pyarrow`/`dask` work out of the box, but `polars` needs `flyteplugins-polars` and LANCE datasets need `flyteplugins-lance` in the image. +4. **Using the pre-2.0 type transformer API** — Flyte 2 transformers use `get_literal_type` / `async to_literal` / `async to_python_value` / `guess_python_type`, not the old `save()`/`load()`/`get_type()` protocol. +5. **Not registering (or distributing) custom transformers** — register via `TypeEngine.register(...)` before execution, or ship them as a `flyte.plugins.types` entry point so they auto-load. +6. **Forgetting `.path` on flyte.io.File/Dir** — inside a task, `file` is an offloaded-type object, not a string. Use `file.path` for the local path and `.name` for the basename.