Calling from a DataFrame

---

You can invoke a function across every row of a dataset using F.catalog_call() inside a static resolver’s DataFrame pipeline. Chalk handles parallelism, batching, and delivery automatically.

For example, define a function that extracts named entities from text:

import chalkcompute

@chalkcompute.function(
    image=chalkcompute.Image.base("python:3.12-slim").run_commands(
        "pip install spacy && python -m spacy download en_core_web_sm",
    ),
    cpu="2",
    memory="4Gi",
)
def extract_entities(memo: str) -> str:
    import json
    import spacy
    nlp = spacy.load("en_core_web_sm")
    doc = nlp(memo)
    return json.dumps([
        {"text": ent.text, "label": ent.label_}
        for ent in doc.ents
    ])

Then apply it across a parquet file of transaction memos with an @offline(static=True) resolver:

import chalkdf
from chalk import offline
from chalk.features import DataFrame, features, _
import chalk.functions as F

@features
class Transaction:
    id: int
    memo: str
    entities: str

@offline(static=True)
def ingest_transaction_entities() -> DataFrame[
    Transaction.id,
    Transaction.memo,
    Transaction.entities,
]:
    return (
        chalkdf.DataFrame.scan(["transactions.parquet"])
        .with_columns({
            "transaction.entities": F.catalog_call("function.extract_entities", _.memo),
        })
        .select("transaction.id", "transaction.memo", "transaction.entities")
    )

Each row’s memo column is passed to extract_entities, and the result is written as the entities feature alongside it. .with_columns() keeps every existing column and adds the ones you name; use .project() instead when you want only the columns you specify. Note the qualified name passed to catalog_call: it’s function.extract_entities, not a bare function name. model.* and function.* are the two valid namespaces.