# What is Chalk? source: https://docs.chalk.ai/docs/what-is-chalk Chalk is the AI/ML data platform, providing infrastructure in your own cloud for: - Getting fresh, contextual data to your models and agents at inference time - Deploying and running those models and agents securely and at scale Chalk offers both the features/context and compute solutions for inference, built on the same philosophy: define what you need in Python, and run it on infrastructure you control. ### Why Chalk? Most teams assemble a production AI stack from a feature store, a vector database, retrieval and prompt tooling, an orchestration layer, a sandbox runtime, and a governance story — then fight to keep them all consistent. Chalk replaces that stack with a single platform: - One definition of your data — features, embeddings, LLM outputs, and prompts are all just Python, computed once and served everywhere: training, real-time inference, and agents. - Point-in-time correct — query any value as it existed at any moment, so training data matches production and agent evals replay exactly what production saw. - Fast in production — features served in single-digit milliseconds, so real-time models and agents never wait on data. - In your own cloud — everything runs in your VPC under IAM roles you control, so your security team evaluates the platform once instead of every agent you ship. ### The Context Engine The Context Engine is a real-time feature computation engine that serves as the single source of truth for your features. You define features once in Python, and Chalk computes and serves them directly from your data sources at inference time in milliseconds . Unlike traditional feature stores, which behave like databases or caching layers, Chalk keeps training and serving perfectly in sync and makes feature development faster. It adds built-in monitoring, a branch-based deployment model for feature versioning, and deployment inside your own cloud. This is especially critical for teams working with sensitive data or in regulated industries. Here's a quick example of declarative feature engineering in Chalk: ``` from chalk.features import features from chalk import online @features class User: id: int total_orders: int chargeback_rate: float fraud_score: float @online def score_fraud( orders: User.total_orders, cbr: User.chargeback_rate, ) -> User.fraud_score: return 0.7 * cbr + 0.3 * (orders < 3) ``` Query the same features in real time, or as of any point in the past for training and evaluation: ``` from chalk.client import ChalkClient # in real time, or as-of any point in the past ChalkClient().query( input={"user.id": 1}, output=["user.fraud_score"] ) ``` ### Chalk Compute Building on that same philosophy, Chalk Compute extends the platform to the context and infrastructure challenges of agents. It provides governed, sandboxed environments where agents securely run and take action on your data — without it ever leaving your VPC. That lets you run scalable agents (or any remote process) quickly, manage data access, and evaluate agent performance against historical time periods. Here's a simple example of spinning up a sandbox: ``` sandbox = Sandbox( cpu="2", memory="4Gi", image=( Image.debian_slim(python_version="3.13").run_commands("pip install claude") ), volumes=[("code", "/code")], ).run() print(sandbox.exec("echo", "hello!").stdout_text) sandbox.terminate() ``` Every sandbox is isolated under gVisor, launches with its own scoped cloud identity, and can be replayed against any point in time. Sandboxes boot in under a second and scale to 10,000 isolated containers in under 10 seconds. The same SDK also runs long-lived model servers — for example, a GPU-backed inference server: ``` import chalkcompute # a GPU-backed model server, in your own cloud server = chalkcompute.Container( image=chalkcompute.Image.base("vllm/vllm-openai:latest"), gpu="nvidia-l4:1", cpu="4", memory="16Gi", port=8000, entrypoint=["--model", "Qwen/Qwen2.5-7B-Instruct"], ) server.run() ``` ### Get started - Install the SDK. pip install chalkpy for the Context Engine, and pip install chalkcompute to run agents and models. - Connect your cloud. Point Chalk at your account so everything runs in your VPC. Chalk Compute runs on AWS or GCP; the Context Engine also supports Azure. - Define your features. Write resolvers for the data your models and agents need, then chalk apply to deploy them. - Query them in real time, or backfill any feature as of any point in the past. - Run your agents. Decorate a Python function with @chalkcompute.function and invoke it remotely. Next steps: read the docs overview, the feature engineering guide, the Chalk Compute overview, and the API reference. # Platform Architecture source: https://docs.chalk.ai/docs/architecture ## How it all fits together. Chalk’s platform is architected to - retrieve fresh features across heterogeneous sources with the minimum possible latency - orchestrate and execute complex transformations on structured and unstructured data - support online serving, offline training-set generation, and streaming ingestion in a single system - uphold enterprise-grade guarantees for security, observability, and reliability In short, Chalk was built to get the right data from the right place at the right time. ### Online serving architecture Let's examine how the pieces work together to compute and serve features. Suppose that you want to compute a set of features for making a decision about a request from a user: - Your application sends an HTTP request to Chalk's serving API for features - Chalk's query planner generates an optimized execution plan by analyzing feature dependencies and available data sources. - Chalk's compute engineretrieves fresh values from underlying sources with Resolverspulls from Chalk’s low-latency online storage e.g. materialized features or cached featuresexecutes the generated query plan - Returns computed features in response - Newly computed feature values are stored and logged for reuse and auditability Chalk architecture diagram This entire online pipeline - from SQL queries and API calls to getting a response - runs in less than 5ms, even with heterogeneous data sources and complex logic. Chalk uses many techniques to reduce latency, such as: - Automatic parallelism across pipeline stages - Vectorization of pipeline stages that are written using scalar syntax - Statistics-informed join planning - Low-latency key/value stores (like Redis, BigTable, or DynamoDB) - Transpilation of Python code into native code with Chalk's symbolic Python interpreter ### Data orchestration Chalk eliminates the complexity of orchestrating data and ETL pipelines by building a dependency graph (DAG) of your features, which are defined using Python. At inference time, Chalk dynamically builds query plans (subgraphs of your feature DAG) without manual configuration, based on the features you request. Write feature definitions in Python, and Chalk automatically - Determines optimal computation and query planning strategies - Handles caching, incremental updates, and backfills - Provides built-in observability and data lineage without additional tooling As a result, Chalk can serve as a drop-in replacement for orchestration tools like Dagster, Airflow, and Prefect while simultaneously providing purpose-built features for production ML workloads. Chalk query plan Declaratively defining features frees up data teams to focus on designing features instead of writing plumbing code. There's no need to write custom glue code because Chalk interfaces directly with underlying data sources, managing all the connections and transformations behind the scenes. Note: Features can also be computed on a recurring basis with scheduled queries or ingested continuously from streaming sources such as Kafka, Kinesis, and Pub/Sub. ### Offline computation and serving Chalk's architecture also supports efficient point-in-time queries to construct model training sets or perform batch offline inference. - Submit a training data request from a notebook client like Jupyter using Chalk's Python SDK - The query planner builds a point-in-time correct plan - The compute engine pulls point-in-time correct feature values from offline storage - Chalk returns a DataFrame of features to you. Chalk integrates with your existing data providers (Snowflake, Delta Lake, or BigQuery) to ingest massive amounts of data from a variety of data sources and query it efficiently. Note that data ingested into the offline store can be trivially made available for use in an online querying context with Chalk's Reverse ETL. There's an exhaustive list of supported ingestion sources in the Integrations section. ### Data persistence and storage Chalk uses different storage technologies to support online and offline use cases. The online store is optimized for serving the latest version of any given feature for any given entity with the minimum possible latency. Chalk can be configured to use Redis or Memorystore for smaller resident data sets with strict latency requirements, or DynamoDB when horizontal scalability is required. The offline store is optimized for storing all historical feature values, serving point-in-time correct queries, and tracking provenance of features. Chalk supports a variety of storage backends depending on data scale and latency requirements. Typically, Chalk uses Snowflake, Delta Lake, BigQuery, Iceberg, or Athena. ### High-performance execution engine Under the hood, Chalk uses Velox, an open source unified execution engine, to deliver high-throughput feature computation. We maintain a fork that's been optimized for low-latency online inference. You can think of Velox as a backend for query engines like Presto (AWS Athena) and Spark i.e. you can't point Velox at a database and pass in a SQL expression. Rather than forcing users to work directly with low-level execution primitives, Chalk provides an ergonomic interface (Chalk Python SDK) for defining features, transformations, and pipelines. Velox architecture diagram This architecture allows us to expose the power of vectorized computation with clean APIs that feel natural (like writing Pandas and Polars) to data scientists and engineers. Users write simple Python decorators and SQL queries, while Velox handles the complex optimizations that make these computations blazingly fast. - Velox's columnar memory layout and vectorized expression evaluation deliver significant performance improvements, with benchmarks showing 6-7x speedups on real analytical workloads - Native support for structs, maps, arrays, and nested data types makes it ideal for feature engineering workflows - Features like filter reordering, dynamic filter pushdown, and adaptive column prefetching optimize query execution based on runtime statistics ### Service architecture We offer both a hosted model (“Chalk Cloud”) and a customer-hosted model (“Customer Cloud”). Most companies choose to run Chalk in their own cloud (VPC) for data residency and compliance. Chalk is traditionally deployed as a Platform As a Service into an isolated account under the management of the customer, with the core components (VPC, Kubernetes Cluster, Message Queues, etc) managed by Chalk. Compute nodes run on Kubernetes (typically EKS on AWS, GKE on GCP, and AKS on Azure). If you have custom needs, we are happy to customize the deployment to fit with your service architecture. ### Metadata Plane The Metadata Plane is responsible for storing and serving non-customer data (like alert and RBAC configurations). It can control many Data Planes, which it manages through the Kubernetes API, enabling tasks such as scaling deployments and running batch jobs. In short, the Metadata Plane handles: - Deployment orchestration - Access control - Monitoring and alerting - Feature discovery It does not have access to customer data. ### Data Plane The Data Plane encompasses the execution environment for feature pipelines along with the storage and serving infrastructure for both online and offline feature stores. A single Data Plane can run many Chalk Environments. Often, companies will have 2-3 environments (like qa, stage, and prod.) If running in a single data plane, these environments share resources, which helps with cost and ease of setup. However, if you prefer to have stronger isolation between Chalk Environments, each Chalk Environment can run in a separate Data Plane. You would typically share one Metadata Plane to orchestrate all Data Planes, and deploy the shared Metadata Plane to the most sensitive of the environments. ### Deployment Models Chalk offers several deployment options to provide the right level of infrastructure control. ### Chalk-Hosted Deployment In the Chalk-Hosted Deployment, both the Metadata Plane and Data Plane run in Chalk's cloud account. Deployed in this manner, Chalk runs as a SaaS application. There is no infrastructure to manage, and no ability to see inside the cloud account running Chalk. ### Customer Cloud Deployment The Customer Cloud Deployment is Chalk's most common deployment. Most customers choose our Customer Cloud Deployment. In this model, the customer runs the Data Plane in their own cloud, while Chalk runs the Metadata Plane in Chalk's managed cloud environment. This deployment model strikes a good balance between security and ease of maintenance. No one at Chalk will be able to access your data, but the Chalk team can handle upgrades to the underlying resources without your team's involvement. ### Air-Gapped Deployment Chalk offers the option to self-host both the Data Plane and Metadata Plane. In the Customer Cloud Deployment, only the Data Plane runs in your cloud account, whereas in the Air-Gapped Deployment, the Metadata Plane joins the Data Plane in your cloud account. There are two primary reasons that customers will choose to host the Metadata Plane: - Regulatory requirements: highly regulated environments (like FedRAMP) may require additional security control. - Disaster recovery: deploying both the Metadata Plane and Data Plane into your cloud account means that Chalk will continue to run independent of any Chalk uptime. In this configuration, no service hosted by Chalk needs to talk to your instance. Telemetry can be exported for billing purposes (over topics), but is non-essential to uptime of your instance. In the event of a complete outage in Chalk's cloud accounts, your instance service would continue running indefinitely without disruption. ### Data Plane Architecture Chalk has a few tiers of concepts within the Data Plane that are important to understand when thinking about how to structure your deployment, such as how many environments you should set up. Each customer is associated with a team. Each team is tied to one or more projects, and each project can have one or more environments. The team, project, and environment concepts are all logical groupings for the underlying Kubernetes resources of Data Plane components. Each environment deploys into its corresponding Kubernetes namespace. An environment has at least one resource group, which is one set of Data Plane components, as well as a metrics database deployment. Several environments can be deployed to the same Kubernetes cluster. Every Chalk-managed Kubernetes cluster has a set of background processes that can be shared across environments deployed to that cluster. These background processes include load balancers, background persistence workers, a Clickhouse deployment for tracing, and a cluster manager. Most customers have a team, a project, and one or more environments deployed across one or more Kubernetes clusters, however the exact structure of your deployment is flexible within the constraints outlined in the previous paragraph. ### Monitoring Native integrations with PagerDuty and Slack ensure teams are immediately alerted to any issues in their feature pipelines. Beyond alerting, every Chalk query is fully instrumented with traces and detailed logs, enabling both broad system-wide monitoring and deep request-level debugging across every stage of computation—down to the root data source. With Chalk, data teams get - Comprehensive logging & debuggingResolver execution detailsFeature computationsCache missesData source failuresFeature stalenessFeature ownership tracking - Metrics and performance monitoringFeatures: request volumes, computation times, and error ratesModels: inference latency, prediction accuracy, and resource utilizationSystem: query throughput, system latency, and pipeline health - End-to-end request tracingTrack the inputs and outputs of every step in any runLineage tracking from data source to final predictions Easily, build your own views and set up custom dashboards to visualize your metrics and configure smart alerts with custom formulas that notify you instantly when thresholds are crossed or anomalies are detected. Metrics and chart creation This flexibility to configure and define your own metrics makes it easy to answer common questions such as how often certain features are computed, how long individual computations take, and what the average value for a feature is. ### Architecting maintainable and scalable systems for enterprise success By both connecting to your data stores directly and computing features post-fetch, Chalk makes it trivial to integrate new data sources from other teams, dramatically increasing predictive accuracy and the context available to your models. Your systems can also bidirectionally integrate with Chalk's underlying infrastructure, which is built on widely-adopted technologies like Redis or DynamoDB, and leverages open standards like Arrow, Parquet, and Iceberg—ultimately maximizing compatibility and unlocking downstream analytical workflows Together, these architectural choices enable enterprises to build future-proof ML and AI systems that scale with their needs, maintain interoperability, and seamlessly integrate with their existing technology stack. # How do I use Chalk? source: https://docs.chalk.ai/docs/setup-guide ## Customizing your Chalk solution. Chalk is a feature store that enables data engineers and data scientists on production machine learning teams to collaborate efficiently and effectively. In this tutorial, we will guide you step by step through customizing your Chalk solution to help you achieve all of your data goals. ### Creating your Chalk project - If you don't already have a GitHub repository where you will store your Chalk code, create one. Then, pick a local directory in which to work on Chalk code, and clone the GitHub repository there. Then cd inside of the directory. - If you haven't already, install Python. You can do this through Homebrew as follows: ``` # Install Homebrew $ /bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)" # Use Homebrew to install python3.11 $ brew install python@3.11 ``` - Install the Chalk command line tool. The Chalk CLI allows you to create, update, and manage your feature pipelines directly from your terminal ``` curl -s -L https://api.chalk.ai/install.sh | sh ``` - Create a Python virtual environment within your repository root directory and activate it ``` $ python3.11 -m venv .venv $ source .venv/bin/activate ``` You can run source .venv/bin/activate to activate the virtual environment, and deactivate to deactivate. - Log in by typing chalk login. If you're using a dedicated environment, make sure you use the --api-host flag. Type y when prompted and log in through the browser. - Run chalk init to initialize your project files. This will initialize a directory structure with an empty src directory and three files: chalk.yaml, README.md, and requirements.txt. ``` root_directory/ ├── src/ ├── chalk.yaml ├── README.md └── requirements.txt ``` The first file, chalk.yaml, stores configuration information about your project. You can edit this so it contains the following ``` project: {YOUR_PROJECT_NAME} environments: default: requirements: requirements.txt runtime: python312 ``` The second file, README.md, contains some basic commands you can use with the Chalk CLI. The third file, requirements.txt should look something like this: ``` requests chalkpy[runtime] ``` This contains your project dependencies. You can add a file .chalkignore which will act just like a .gitignore file, and exclude the specified files from deploys when you run chalk apply (more on this later). - Within your virtual environment, install the project requirements. ``` $ pip3 install -r requirements.txt ``` Now, you should have a virtual environment with all project dependencies installed and a basic project structure upon which we will build in the next step. ### Configure data sources Having set up a basic project directory, next we'll want to configure the data sources from which we will load data to compute our features. We can do so in the Chalk dashboard. - In the same directory from before, log in or sign up with Chalk directly from the command line. The chalk login command will open your browser and create an API token for your local development, as well as redirect you to your dashboard. If you are not redirected you can also find the dashboard at https://chalk.ai/projects or run the chalk dashboard command. You will automatically see all the environments in which the email you used to log in has been provisioned as a user. - Within the dashboard, navigate to Data Sources in the sidebar and add all data sources that you will be working with here. After you have saved a data source, you can use the Test Data Source button in the upper right hand corner of the Data Source configuration view to verify that your connection is valid. - Within the working directory, we'll add a datasources.py file under our src folder to reference the data sources that we've added in the dashboard. ``` root_directory/ ├── src/ │ ├── __init__.py │ └── datasources.py ├── chalk.yaml ├── README.md └── requirements.txt ``` Say we added a PostgreSQL data source, then our datasources.py might look something like this: ``` pg = PostgreSQLSource(name='PG') ``` For more details on setting up data sources, see here. ### Define feature classes and resolvers Next, we'll define our feature classes and resolvers. Each feature class is a Python class of features, and each resolver tells Chalk how to compute the values for different features. Each feature that we write should correspond to a resolver output. We recommend starting with a minimal feature class and building up iteratively to easily test your code along the way. After writing some feature classes, resolvers, and tests, we would expect to see a directory structure like this: ``` root_directory/ ├── src/ │ ├── resolvers/ │ │ ├── .../ │ │ ├── __init__.py │ │ └── pipelines.py │ ├── __init__.py │ ├── datasources.py │ └── feature_sets.py ├── tests/ │ └── ... ├── .chalkignore ├── chalk.yaml ├── README.md └── requirements.txt ``` You can read more in our docs about the different kinds of features and the different kinds of resolvers that you can write. If you would like guidance on how to structure your feature classes and resolvers, please reach out in your support channel! ### Deploy and query Now, you can deploy the features and resolvers that you wrote! You can deploy to production by using the chalk apply command. During development, we recommend that you use chalk apply --branch {BRANCH_NAME} to deploy to the branch server, which allows multiple people to work concurrently in one environment, and also enables more performant deploys. Once you have deployed your code, then you can query your features directly from the command line using the chalk query command, or by calling one of our Chalk Clients in code. Chalk has a Python client, Go client, Java client, and a Typescript client. This is the primary workflow for iterating on features and resolvers! Write, deploy, and query to verify whether the feature values that you receive are the values you expect. Once you have finalized your feature class and resolver definitions, the final step is frequently orchestration. ### Orchestration Having verified that your feature and resolver definitions are correct, the next step is to determine how you want to use the corresponding feature values within your larger machine learning platform. Some users trigger resolvers and run queries from within other orchestrated pipelines, such as Airflow. Some users define cron schedulesfor their resolvers, and set staleness values on different features to ensure that the data they query falls within their requirements for freshness. The data world is your oyster! But, as always, if you would like guidance on how to configure that oyster, please reach out in your support channel! ### Further resources For a detailed tutorial on how to build a fraud model using Chalk, see here. # File Structure source: https://docs.chalk.ai/docs/configuration ## Configure your Chalk project and organize your features and resolvers. ### Overview Your Chalk project's configuration is shared across the following files: - chalk.yaml (or chalk.yml): Configuration for your project's deployment - .chalkignore: Files to exclude from your project's deployment - A requirements file: Default Python requirements for Chalk to install via pip (can be overridden in chalk.yaml). This can either be requirements.txt or a pyproject.toml (via Poetry or uv). Here's our recommended repository structure: ``` company_chalk/ ├── src/ │ ├── resolvers/ │ │ ├── ... │ │ ├── __init__.py │ │ └── pipelines.py │ ├── __init__.py │ ├── datasources.py │ └── feature_sets.py ├── tests/ │ └── ... ├── notebooks/ │ └── ... ├── .chalkignore ├── chalk.yaml ├── README.md └── requirements.txt ``` When you're first getting started, we recommend putting all your features in a single file. Keeping the features in a single file makes circular references easier to reason about, as they can just be quoted. If you do want to split your features across multiple files, you'll need to use the if TYPE_CHECKING block from the typing module. ### Organizing Feature Definitions Across Files In large projects, it's common to split feature definitions across multiple Python modules. For unidirectional dependencies, this is straightforward. For example, if src/models/user.py imports src/models/profile.py, you can define the User and Profile features in separate files without issues. However, if you have circular dependencies, you may run into problems. Chalk supports this, but circular imports can arise when features reference each other across files. To avoid these issues, use the if TYPE_CHECKING block from the typing module and quote your forward references. Here's an example of how to do this cleanly: ``` # Imports `User` directly, because `src/models/user.py` # doesn't import `src/models/profile.py` from chalk.features import features from src.models.user import User @features class Profile: id: Primary[User.id] username: str ``` ``` from chalk.features import features from typing import TYPE_CHECKING if TYPE_CHECKING: # Imports `Profile` only when type checking # to avoid circular imports from src.models.profile import Profile @features class User: id: str # Profile must be quoted because it is imported # only when type checking profile: "Profile" ``` By quoting imports inside if TYPE_CHECKING, you avoid circular dependency errors while still maintaining type safety and feature linkage. If the relationship to Profile is optional, you can use typing.Optional or the | None syntax, but the entire annotation should be quoted: ``` from chalk.features import features from typing import TYPE_CHECKING if TYPE_CHECKING: from src.models.profile import Profile @features class User: id: str # All of `"Profile | None"` must be quoted, not just the `Profile` part profile: "Profile | None" ``` A similar pattern should be used for DataFrame annotations: ``` from chalk.features import features from typing import TYPE_CHECKING if TYPE_CHECKING: from src.models.profile import Profile @features class User: id: str # All of `"DataFrame[Transaction]"` must be quoted, not just the `Transaction` part transactions: "DataFrame[Transaction]" ``` In a project with features living in different files, we recommend that the schema definition all live in a single folder, separate from the resolvers. In this case, your folder structure will look something like: ``` company_chalk/ ├── src/ │ ├── models/ │ │ ├── user.py │ │ ├── profile.py │ │ ├── __init__.py │ │ └── ... │ ├── resolvers/ │ │ ├── ... │ │ ├── __init__.py │ │ └── pipelines.py │ ├── __init__.py │ ├── datasources.py │ └── feature_sets.py ├── tests/ │ └── ... ├── notebooks/ │ └── ... ├── .chalkignore ├── chalk.yaml ├── README.md └── requirements.txt ``` Keeping the schema definition all in one place (as you might with something like Protobuf or Avro files) helps you keep your schema definitions organized and makes importing them in your resolvers straightforward. ### chalk.yaml Use chalk.yaml to configure your project's Docker environment, Python configuration, and metadata validation for your features and resolvers. ### Schema The name of this Chalk project, which must match your project's name in the dashboard (case-sensitive). Per-environment overrides. See our Docker documentation for more details. The name of this environment, such as prod or qa. Find available environment names in your dashboard. Use default to configure default values to apply to all of this project's environments. The Python version for the Chalk project. Python requirements file to install with pip. This can either be a requirements.txt file or a pyproject.toml file. Path to an alternative Dockerfile that will be used to build a base image for your Chalk deploys. Reference a version of the Chalk runtime that you want to use for this environment. Can be either a version tag, like v3.0.0, or a release stream like stable. If not specified, defaults to stable. Releases are documented in the Platform Release Notes. Path to a custom ignore file for this environment. If not specified, defaults to .chalkignore in the project root. Use this to exclude different files from deployment in different environments. Configuration for metadata validation for features and resolvers. See our validation documentation for more details. Validations for all features. A list of dictionaries of metadata attributes and error severity levels when this metadata property is missing. The name of the metadata property. The level of severity to use when a feature is missing this metadata property. Deploys are permitted for info and warning levels, but disallowed for error level. Validations for all resolvers. A list of dictionaries of metadata attributes and error severity levels when this metadata property is missing. The name of the metadata property. The level of severity to use when a feature is missing this metadata property. Deploys are permitted for info and warning levels, but disallowed for error level. ### Sample file Here is a sample chalk.yaml file. In this file, we use a different Dockerfile in production and different chalkignore files per environment. ``` project: my-project-id environments: default: runtime: python312 requirements: requirements.txt dev: chalkignore: .chalkignore.dev prod: dockerfile: ./DockerfileProd chalkignore: .chalkignore.prod validation: feature: metadata: - name: owner missing: error - name: description missing: warning - name: tags missing: info resolver: metadata: - name: owner missing: error ``` ### .chalkignore Your .chalkignore file should include your scripts, notebooks, and tests. Anything that you are not actively using in your deployment should be added so that non-deployment code does not clutter or interfere with your deployment. ### Per-Environment Chalkignore Files You can specify different chalkignore files for each environment in your chalk.yaml file. This is useful when you want to exclude different files from deployment in different environments. For example, you might want to include experimental features in dev but exclude them from production. ``` project: my-project environments: dev: runtime: python312 chalkignore: .chalkignore.dev # Custom ignore file for dev prod: runtime: python312 chalkignore: .chalkignore.prod # Custom ignore file for prod staging: runtime: python312 # No chalkignore specified - defaults to .chalkignore ``` If no chalkignore field is specified for an environment, Chalk will use the default .chalkignore file in your project root. ### Python Requirements Chalk will install the Python requirements for your project as specified in the requirements parameter of your chalk.yaml file. This can either be a requirements.txt file or a pyproject.toml. Chalk supports both Poetry and uv for managing your Python dependencies. You can specify this file's location and type in your chalk.yaml file like below: ``` project: my-project-id environments: dev: runtime: python312 requirements: requirements-dev.txt prod: requirements: pyproject.yaml ``` ### Python Requirement Overrides The requirements file specified in your chalk.yaml file will serve as the default dependency set for your environment. However, if you have different requirements for specific resolvers, you can also define alternative virtual environments under the venvs key in your chalk.yaml file. ``` project: features environments: default: runtime: python311 requirements: ./requirements.txt platform_version: v3.26.8 venvs: model-1: dependencies: - numpy==2.3.2 - pandas==2.3.2 - scikit-learn==1.7.2 - tensorflow==2.20.0 - torch - transformers model-2: requirements: ./requirements-model-2.txt ``` For each virtual environment, you can specify the associated dependencies either by passing in a list of Python requirements under the dependencies key, or by providing a filepath to a requirements.txt or pyproject.toml file under the requirements key. Then, to run a specific resolver in one of these virtual environments, you can reference your virtual environment using the name specified in your chalk.yaml file in the @before_all, @online, and @offline decorators. For your @before_all functions, if there is no venv specified, the function will run in every virtual environment. Otherwise, if there is a venv specified, the function will only run in the specified virtual environment. Otherwise, @online and @offline resolvers will default to run using the default virtual environment, unless a different venv is specified. ``` from chalk import before_all, online, offline, Features from src.models import User # this will run in all venvs @before_all() def before_all_hook(): from services import msvc msvc_client = msvc.Client() # runs **only** in specified venv @before_all(venv="model-1") def before_all_model_1(): from models import Model1 model_1 = Model1.load() # runs **only** in specified venv @before_all(venv="model-2") def before_all_model_2(): from models import Model2 model_2 = Model2.load() # runs using default venv @online def get_user_email_domain( email: User.email ) -> Features[User.email_handle, User.email_domain]: handle, domain = email.split("@") return User(email_handle=handle, email_domain=domain) # runs using model-1 venv @online(venv="model-1") def get_user_risk_score_1( id: User.id, email_domain: User.email_domain, recent_activity_count: User.recent_activity_count["7d"] ) ->User.is_fraud_1: threshold = msvc_client.get_model_threshold("model-1") return model_1.predict(id, email_domain, recent_activity_count).sum() > threshold # runs using model-2 venv @online(venv="model-2") def get_user_risk_score_2( id: User.id, email_domain: User.email_domain, recent_activity_count: User.recent_activity_count["7d"] ) -> User.is_fraud_2: threshold = msvc_client.get_model_threshold("model-2") return model_2.predict(id, email_domain, recent_activity_count).sum() > threshold ``` ### Pinning Chalk Platform Versions By default, when you deploy in your Chalk environment, Chalk will pull the latest version of the Chalk platform. You can choose to pin a Chalk platform version for each environment such that your deployments will only update your Chalk code, and will use the same platform image until you deploy a new pinned platform version. ``` project: my-project-id environments: dev: runtime: python312 requirements: requirements-dev.txt platform_version: v3.27.30 staging: runtime: python312 requirements: requirements-staging.txt platform_version: v3.27.30 prod: runtime: python312 requirements: requirements.txt platform_version: v3.27.30 ``` ### The Chalk Runtime Package chalkruntime is the underlying library that drives Chalk's feature processing — it manages the core mechanics of your environment. While it isn't intended for standalone use, it's useful to know about if you need precise control over dependency versions. Including chalkruntime in your requirements lets you inspect the exact package versions running in your live environment and keep your local setup in sync with the remote platform configuration. Pick the chalkruntime version that matches your pinned platform version — available versions are listed on PyPI. # Best Practices source: https://docs.chalk.ai/docs/best-practices ## Learn our best practices for building and maintaining your Chalk solution With Chalk, the same solution can be implemented in a number of different ways. Below are some guidelines and recommended patterns for building and maintaining your Chalk solution! ### Data Sources/Integrations ### Test your data source connections Define integrations through the Chalk dashboard and check that they're connecting properly using the Test Data Source button. Test Data Source ### Avoid naming your data sources by their "type", e.g. don't call your Postgres data source "postgres" Though this works, it can lead to ambiguity in SQL resolvers when you add multiple data sources of the same type. This occurs because Chalk lets you refer to data sources by their type if you've only linked one data source of that type to Chalk. In general, to future-proof your resolvers, you should refer to them by their name. ### Features ### Start by defining your features Features fully specify what you want your data to look like. Once you have an understanding of the inner relations, writing resolvers for your features becomes easier. ### Avoid dataclass feature types, instead use separate feature classes or unpack the data While Chalk allows for dataclass feature types, they should be avoided. They don't always play nicely with serialization and can cause tough-to-debug errors. We recommend either unpacking the nested class into basic types or defining and joining an additional Chalk feature class if the nested component is truly a separate entity. Defining a separate feature class also makes the underlying data easier to monitor and test. ### Tag and annotate your features We strongly recommend annotating your features as you are developing them. Your feature annotations show up in the Chalk dashboard and are a good way of documenting your code. You can also add tags and owners to your features, which can be used for aggregation and filtering in the Chalk Dashboard. ``` from chalk.features import features from datetime import datetime @features class User: id: str # the user's fullname # :owner: mary.shelley@aol.com # :tags: team:identity, priority:high name: str # the user's birthdate # :owner: mary.shelley@aol.com # :tags: team:identity, priority:high birthday: datetime ``` If you want to apply a tag or owner to all features in a feature class, this should be done in the feature decorator, like so: ``` from chalk.features import features @features(owner="ada.lovelace@aol.com", tags=['group:risk']) class User: id: str # the user's fullname. name: str ``` You can also apply restrictions or enforce feature annotation for your entire project. For instance, you can block deployment if features are not tagged or described. ### Centrally manage feature-permission tags Feature tags drive feature permissions, so the set of tags on your features is your access-control policy. Declaring tags inline next to each feature spreads that policy across your entire codebase — a large surface to audit, and an easy place for a tag to be added or changed without review. override_feature_tags collapses that surface to a single file. Calling it switches your project into override mode: at deploy time every code-defined tag (inline feature(tags=...), class-level @features(tags=...), and # :tags: comments) is discarded, and the only tags that apply are the ones in the central mapping. ``` # feature_tags.py — the one place that defines feature access from chalk.features import override_feature_tags override_feature_tags( { Transaction.raw_amount: ["pii"], Transaction.amount_bucket: ["cleared"], User.email: ["pii"], } ) ``` Because this becomes the only source of tags, a reviewer has exactly one file to read to reason about who can access which features. Pair it with a CODEOWNERS rule requiring sign-off from your security or platform team on that file, and the organization gets a single, reviewed chokepoint for feature access. Without it, any developer could grant a sensitive feature an AllowDownstream tag from any file and self-escalate access. The override is applied over the fully-populated feature registry at deploy time, so it is independent of import order, and you can split the mapping across multiple override_feature_tags(...) calls within that one reviewed file. ### Keep feature definitions separated from resolvers Features should be defined in separate files from resolvers (except underscore features). ### When starting your Chalk implementation, define all your feature classes in the same file Although it can get a bit lengthy, we recommend starting by defining your features in a single file. This makes expressing joins between features easier and prevents circular dependencies. ### Add Validations For Your Features Validations for your features can prevent incorrect data from being written to your offline store. They can provide an even stricter complement to monitoring, ensuring that nothing is going wrong with the feature you are calculating. ### Use implicit join syntax Joins between feature classes can be specified in a number of different ways. We recommend using an implicit join syntax, which we cover in the join section of the docs. ### Resolvers ### Use SQL resolvers to read data from your raw datasets While Python resolvers can be used to read data from your data sources, SQL resolvers are preferred. SQL resolvers allow for direct execution against your data sources and additional optimizations, making them more efficient. ### Prefer Chalk expressions over Python resolvers for simple computations ``` # Preferred: Chalk expression @features class User: id: str age: int is_adult: bool = feature(expression=_.age >= 18) # Avoid: Python resolver for simple computation @online def get_is_adult(age: User.age) -> User.is_adult: return age >= 18 ``` ### Explicitly list columns in a select statement for SQL file resolvers Select statements in SQL resolvers should be explicit. Avoid using the * syntax. ### Give Python resolvers and SQL filenames clear names Resolver names are used in the Chalk dashboard to identify resolvers. They should be clear and concise. ### Make sure your resolvers operate inside a single feature space Resolver inputs and outputs must belong to the same feature class, but joins can allow resolvers to connect data between feature classes. ### Transform your Chalk DataFrame to Pandas or Polars Don't worry about converting Chalk DataFrames to Pandas or Polars in a Python resolver—the transformation is cheap. We use arrow (and so do Pandas and Polars) so moving data from a Chalk DataFrame to either is close to free ### Querying ### Run simple queries with the Chalk CLI, for more flexibility use one of Chalk's API clients The Chalk CLI should be used to run simple online queries. For more complex use cases, you should use one of Chalk's API clients. ### Run large queries asynchronously When sending large or long-running queries, use run_asynchronously to set up one Kubernetes pod per input, separate from your production and branch servers. You can configure resources for asynchronous offline queries in your dashboard settings. You can also terminate asynchronous queries through the dashboard by terminating the pod or canceling the query within the dashboard, which is not possible for synchronous queries. ### Create named queries for your commonly executed queries With Chalk, you can alias your complex queries using a NamedQuery. Named queries help simplify and document your query patterns. To define a named query, add a NamedQuery object to your Chalk deployment: ``` from chalk import NamedQuery from src.models import User NamedQuery( name="fraud", input=[User.id], output=[ User.email_age_days, User.denylisted, User.credit_report.flags, ], tags=["team:fraud"], staleness={ User.credit_report.flags: "30d" }, owner="jodie@chalk.ai", description="Primary fraud model for signup", ) ``` Running chalk apply makes the query name available as an alias for executing your more complex queries. For instance, running: ``` chalk query --in user.id=1 --query-name fraud ``` is equivalent to running the more complicated: ``` chalk query \ --in user.id=1 \ --out user.email_age_days \ --out user.denylisted \ --out user.credit_report.flags \ --staleness user.credit_report.flags=30d \ --tag team:fraud ``` This feature is also accessible in all of our API clients through the query_name parameter. For instance, in python, you can run: ``` from chalk.client import ChalkClient ChalkClient().query( input={"user.id": 1}, query_name="fraud", ) ``` To see all the named queries you've defined in your current active deployment, you can run: ``` chalk named-query list ``` If you want to create multiple versions of a similar queries, you can use the version parameter of the NamedQuery object and the query_name_version parameter of our various clients. ### Deployment ### Customize your Chalk project with configuration files Chalk expects to find a chalk.yaml file in your project's root repository. This file stores configuration for your deployment (such as your Dockerfile and Python requirements). You can find a sample Chalk project repository structure and more details in our configuration documentation. You'll also find details on how to exclude files from your deployment with .chalkignore. ### Code changes should be tested and queried on the branch server. Use the branch server to test that your deployments and new features are behaving as expected. ### Use the @before_all decorator to configure global variables for your resolvers Global setup for resolvers should be done through a function decorated with the @before_all decorator. This also allows for unique setups for different environments. ### Custom files such as machine learning models are accessed with the TARGET_ROOT environment variable To access files packaged in your chalk deployments, use the TARGET_ROOT environment variable to fully specify the path to your files. For instance, if you have the following directory which you are deploying to Chalk: ``` example/ ├── chalk.yaml ├── features.py ├── model.joblib └── resolvers.py ``` You would access the model.joblib file as follows: ``` import os model_file_path=f"{os.environ['TARGET_ROOT']}/model.joblib" ``` ### Observability / Testing ### Set up monitoring on your most important features and resolvers early Monitoring helps you catch tricky bugs early and gives you guarantees about the data you are generating and serving. You should configure monitoring for your important features and resolvers early in your implementation. ### Unit test your resolvers to make sure they're functioning as expected Chalk makes it easy to set up unit tests for your resolvers using Pytest or any other python testing framework. ### Improve CI/CD in your Chalk repository using our GitHub Actions workflow Chalk has a GitHub Actions integration—you can use it to create branch deployments or run queries as part of your code development cycle. ### Security ### Generate and use access tokens to set and restrict the permissions for your different users Scope your Chalk user permissions with access tokens. These can be programmatically generated through Chalk clients or in the dashboard. Give your users only the permissions they need. # Frequently Asked Questions source: https://docs.chalk.ai/docs/faq ## A collection of questions frequently asked by customers ### Getting support The questions are sorted by category. Don't see your question answered? Please reach out via your support channel and we will be happy to help! ### Data Sources and Infrastructure ### What is the difference between the online store and the offline store? The online store is intended to store features for low-latency retrieval in online query. Typically, the online store is implemented using Redis or DynamoDB. The offline store is intended to store historical logs of all previously ingested or computed features. It is used to compute large historical training sets. It is typically implemented using BigQuery, Snowflake, Iceberg, or other data warehouses. ### Can we do RBAC (Role Based Access Control) within Chalk? Yes! Within the dashboard you can assign roles with different permissions to different users. The default roles available are shown below. ### What are the necessary steps for us to get Chalk in our system? Please reach out via your support channel and we'd be happy to walk you through how to get Chalk setup running on your cloud infrastructure! ### We use Okta SCIM provisioning. Can we import these roles to be used in Chalk? Yes! You can set up Okta to automatically provision and deprovision Chalk users. For more information on how, see here. ### How should I set up secrets to be accessible in our deployed environments? You can configure secrets as environment variables in the Secrets tab of Settings in your Chalk dashboard, and these secrets would then be available on the next deploy in the environments that you've specified, and on all branches in those environments! ### Features ### Does Chalk have a feature catalog? Yes! You can view all the features for all namespaces deployed in your environments, along with some metadata on recent activity and updates. ### How should I use feature versioning? You can define versions for features if, for example, you are iterating on feature definitions. We'd recommend setting a default feature version for ensuring consistency across feature references. For more information, please see this tutorial on feature versions ### How does FeatureTime work? Should I override FeatureTime with timestamps from existing data sources? Feature time is the time at which a feature was observed. By default, feature time is set to the feature's resolver execution time. You can override feature time and may want to do so if you're ingesting historical data. For more information on how feature times work, see our time documentation. ### How should I structure my windowed features? What kind of windows can I use? We recommend writing windowed features for aggregated computations over different time periods. You can define the different windows in terms of weeks, days, hours, minutes, or even seconds (weeks="w", days="d", hours="h", minutes="m", seconds="s"). Under the hood, we normalize the time representations into durations in seconds, so you would have multiple syntactical options for accessing different windows for features. We recommend setting default values for windowed features in case there are windows with no events during the specified time period. Otherwise, you can write resolvers defining how your windowed features should be computed and pass windowed features as inputs into resolvers just like normal features! ### Can I upload features into the online store with an API endpoint? Yes! In addition to streaming and scheduled bulk ingests of features, you can submit requests using the upload_features SDK endpoints to synchronously ingest features into the online or offline stores using API clients. ### Resolvers ### What is the difference between an online and offline resolver? Not much! The only difference is in which contexts the resolvers are executed. A few key scenarios: - in "online query" (i.e. queries submitted via .query, .query_bulk, multi_query), offline resolvers never execute. - in "offline query" (i.e. queries submitted via .offline_query), offline resolvers are preferred, taking precedence over online resolvers that compute the same features. - in offline query, online resolvers are permitted to execute @offline is intended to be used for resolvers whose backing data sources are too slow or expensive to fulfill online query requests -- i.e. data warehouses or certain API sources, but if your query requests can tolerate the latency of one of these slow sources, you can mark resolvers using those datasources @online to query them on-the-fly. On the other hand, if your query requests can't tolerate the latency of the underlying data store, evaluate using scheduled ingestion or streaming resolvers in order to ensure that fresh data is available in the online store. ### How do I force certain resolvers to execute in my query? Make use of tags=[...]. If a resolver is marked with a tag, it is only eligible to execute in queries that have a matching tag. An even stricter variant of this concept is available with the required_resolver_tags argument to query and offline_query which allows you to force all resolvers to have a particular tag. ### How do I know when my resolver will run and when it will time out? You can set cron schedules for your resolvers, which will be parsed in your local system's timezone. You can also set customized timeouts for resolvers. Chalk currently has a max timeout of 18 hours for resolvers, but if this does not suit your needs, please reach out to us in your support channel with a description of your compute needs for your resolver! ### How do I know if my resolver is still running? The best way to check if your resolver is still running is to see if there are still resources provisioned to run your resolver. For scheduled resolver runs, you can view an overview of Run History as well as a view of Cloud Resources dedicated to scheduled resolver runs under the Runs tab on the menu sidebar. For triggered resolver runs, you can also find which pod is running your resolver on the run page. Finally, if you navigate to Infrastructure > Resource Configuration, you can view all pods currently running in your cluster, whether for resolver runs or queries. ### How should I work with dataframes in my resolvers? You can define resolvers with DataFrame inputs or outputs, which uses the Chalk DataFrame structure. You can do some aggregations and projections using the Chalk DataFrame, but you can also convert the Chalk DataFrames into Pandas or Polars using .to_pandas() and .to_polars() for more data manipulation. ### Querying ### Can I query for two different feature classes in a single query? We recommend creating a "root" feature class that models the interaction between the two entities -- i.e. if need "customer" and "business" features for a transaction fraud model, you might create a class like this: ``` @features class AuthQuery: # A unique ID that represents this interaction; may be randomly generated if you have # no natural ID in your system. id: str # A reference to the relevant customer. customer_id: "Customer.id" customer: Customer # A reference to the relevant business. business_id: "Business.id" business: Business ``` Then, queries can be submitted using: ``` ChalkClient().query( input={ AuthQuery.id: ..., AuthQuery.customer_id: ..., AuthQuery.business_id: ..., }, output=[ AuthQuery.customer, AuthQuery.business, ] ) ``` ### How do I query for multiple entities at the same time? Use the .query_bulk SDK method instead of .query. For example, to query features for multiple Hospital entities, you might use: ``` result = ChalkClient().query_bulk( input={Hospital.id: [1,2,3,4,5]} output=[Hospital.current_waiting_time, Hospital.has_trauma_bay, Hospital.has_er, Hospital.has_mri] ) df = result[0].to_pandas() ``` ### How do I tell if my offline query is still running if it shows as In Progress in the UI? The first thing to check is whether there is a pod in your cluster running your offline query. You can verify this by viewing the pods running under Infrastructure. If you're running an offline query from a notebook, then the polling may timeout even if the offline query is still running, so the best way to verify the status of your offline query is a combination of checking the query run in dashboard and the status of pods in the cluster. For more visibility, you can also add the run_asynchronously=True argument to ChalkClient.offline_query to explicitly run your offline query on an isolated worker so you can use the worker status as a query status. ### How do I sample instances of a feature class? You can query for random instances of a feature class by running an offline query and specifying the max_samples parameter. For instance, running the example query below will sample ten random users from the User feature class, returning the features specified by the output parameters: [User.id, User.name, User.age]. ``` from chalk.client import ChalkClient from datetime import datetime from src.feature_sets import User client = ChalkClient() dataset = client.offline_query( output=[User.id, User.name, User.age], max_samples=10, recompute_features=True ) df = dataset.get_data_as_pandas() ``` ### Deployment ### How are resources provisioned for my Chalk cluster, and can I modify the configuration? We have default resource configurations for general environments. You can modify the configuration for your project's cloud resources by modifying the specs under Infrastructure > Resource Configuration. You must hit Save and Apply Changes in order for your configuration changes to go through. If you are not sure how you should configure your cloud resources, please reach out to us in your support channel! ### Can I suspend or delete branches when they are not in use? You cannot currently suspend or delete branches, but stale branches do not consume resources, so there is no cost or performance impact from old branches. ### Observability and Testing ### Is my (resolver / query) running? See answers above under Resolvers and Querying respectively! ### How do I setup a sensor so that I can run something else after a resolver run? If you are triggering a resolver run as part of an orchestrated pipeline, we usually see customers using the built-in sensors from the orchestrators (e.g. Airflow or Dagster) to poll for resolver completion. You can also customize this polling using an API to query run status. See here for an example of how to set up Airflow orchestration to trigger and poll a resolver run. ### Roadmap ### What have the people at Chalk been working on? To get a glimpse of recently released features, take a look at our changelog! ### I want to request a feature! Feel free to reach out to us with feature requests or questions about feature requests in your support channel! # Features Overview source: https://docs.chalk.ai/docs/features ## Define features for training and inference. Chalk lets you spell out your features directly in Python. Features are namespaced to a FeatureSet. To create a new FeatureSet, apply the @features decorator to a Python class with typed attributes. A FeatureSet is constructed and functions much like Python's own dataclass. ### Example ``` from datetime import datetime from typing import Optional from chalk.features import features @features class User: id: int full_name: str nickname: Optional[str] email: Optional[str] birthday: datetime fraud_score: float ``` ### Namespacing Features are namespaced by their containing FeatureSet, and then by the name of the variable. In the above example, our features, when rendered as strings, are: | Feature Name | Type | | ---------------- | -------------- | | user.id | Integer | | user.full_name | String | | user.nickname | String \| None | | user.email | String \| None | | user.birthday | Datetime | | user.fraud_score | Decimal | (FeatureSet names are stripped of the suffix "Features", if it exists). ### Overrides Feature names and feature classes can be overridden by supplying the name keyword argument to the feature function or the @features decorator. This practice allows us to evolve our variable names without losing the past history of this feature. Use @features(name=...) to keep the same feature namespace while renaming the Python class: ``` - @features + @features(name="prince") - class Prince: + class TheArtistFormerlyKnownAsPrince: birthday: datetime ``` Use feature(name=...) to keep the same feature name while renaming the Python attribute: ``` from datetime import datetime from chalk.features import feature, features @features class User: - birthday: datetime + date_of_birth: datetime = feature(name="birthday") ``` ### Primary keys Feature sets must all have a primary key. This primary key is used to index storage for features you later resolve in this feature class. Your primary key can have type int or str, given by the type annotation on the field. By default, if you have a feature with the name id, that feature will be the primary key. However, you can override this behavior: ``` from chalk.features import features, Primary @features class User: user_id: Primary[str] ... ``` If you mark an explicit primary key, it will override the default behavior: ``` @features class User: user_id: Primary[str] # Not really the primary key! id: str ``` Alternatively, you can use the feature function to set a feature to primary: ``` from chalk.features import features, feature @features class User: user_id: str = feature(primary=True) ``` ### Versions Chalk versions all of your features with every deployment. However, you can also choose explicit versions for your features. ``` @features class User: ... email_domain: str = feature(version=2) ``` ### Feature time By default, Chalk marks the time a feature was created as the time that its resolver was run. However, you may want to provide a custom value for this time for data sources like events tables. You can inspect the time a feature was created and set the time for when a feature was created by creating a feature of type FeatureTime. To set the time a feature was created, assign the feature when you resolve it: ``` @offline def fn(uid: User.uuid) -> Features[User.name, User.ts]: return User( name="Anousheh Ansari", ts=datetime(month=9, day=12, year=1966) ) ``` Then, when you sample offline data, the name feature will be treated as having been created at the provided date. ### Constructing feature classes To construct a User instance, supply the feature values to the __init__() method ``` User(full_name="Grace Hopper", nickname="Amazing Grace") User(email="grace.hopper@yale.edu") ``` The @features decorator adds a custom __init__(): ``` def __init__( self, uid: int | MISSING = MISSING, full_name: str | MISSING = MISSING, email: Optional[str] | MISSING = MISSING, ... ): self.uid = uid self.full_name = full_name self.email = email ... ``` Note that all fields have a default MISSING value. Therefore, you can construct feature classes with any subset of the fields you would like to use. ### Refactoring After going to production, you may find that you want to change the name of a property on the feature class. You can change the name of a feature property without changing the underlying data using the name override. From the example in the namespacing section, if you initially called a feature birthday, and decided to rename it date_of_birth, you can keep the underlying data the same and rename the property on the class as follows: If you rename the feature class itself, keep the namespace stable with @features(name=...): ### Interplay with auto-id Where the name of the Python property and the name provided to feature(name=...) differ, IDs are auto-assigned based on the name provided to feature(name=...). ### Default feature values For features that can't always be computed, you can pass default to feature or assign a default directly: ``` from chalk.features import features @features class User: id: int + num_purchases: int = 0 + count_logins: int = feature(default=0) ``` # Feature Types source: https://docs.chalk.ai/docs/feature-types ## Define features for training and inference. ### Scalars Features can be any primitive Python type: ``` from enum import Enum class Genre(Enum): FICTION = "FICTION" NONFICTION = "NONFICTION" DRAMA = "DRAMA" POETRY = "POETRY" @features class Book: id: int name: str publish_date: date copyright_ended_at: datetime | None genre: Genre ``` In addition to the primitive Python types (int, float, str, bool), Chalk also supports date, datetime, and the Chalk FeatureTime as scalar feature types. ### Lists and Sets Features can be a list or set of scalar or struct types. ``` @dataclass class Chapter: start_page: int end_page: int @features class Book: authors: list[str] categories: set[str] chapters: list[Chapter] ``` ### Structs Features can be a struct, which is a collection that maps a fixed set of keys to sub-fields. You can use dataclasses, Pydantic models, and attrs classes to represent struct features. Struct features can be used recursively within list features or other struct features. Struct types should be used for objects that don't have ids. If an object has an id, consider using has-one. ### Dataclass You can use any dataclass as a struct feature. ``` @dataclass class JacketInfo: title: str subtitle: str body: str @features class Book: id: int jacket_info: JacketInfo ``` ### Pydantic models Pydantic models can also be used for structs. ``` from pydantic import BaseModel, constr class TitleInfo(BaseModel): heading: constr(min_length=2) subheading: Optional[str] @features class Book: title: TitleInfo ... ``` ### Attrs Alternatively, you can use attrs. ``` import attrs @attrs.define class TableOfContentsItem: foo: str bar: int @features class Book: table_of_contents: list[TableOfContentsItem] ... ``` ### Document Both dataclass and Pydantic structs are implemented using the PyArrow serialization format, a high-performance schema for data serialization. This data is stored "value only", i.e. without keys, so any change to these structs over time will invalidate historical data. To support feature values where the schema changes over time, we introduced the Document struct type. Documents are serialized as JSON and supports changes to schema over time, at the cost of a small performance penalty. ``` from pydantic import BaseModel from chalk import Document class AuthorInfo(BaseModel): first_name: str last_name: str @features class Book: title: Document[AuthorInfo] ... ``` ### Vectors Features can be vectors (fixed sized arrays of floats), such as the output from an embedding model. Unlike list or set features, vector features are compatible with embedding functions and nearest neighbor similarity search. ``` from chalk.features import features, Vector @features class Document: embedding: Vector[1536] # Defines a vector with 1536 dimensions ``` When using the built-in embedding functions, then the vector dimensions don't need to be specified, as Chalk will automatically infer it from the embedding model. ``` from chalk.features import embed, features, Vector @features class Document: content: str # Chalk knows that text-embedding-ada-002 has 1536 dimensions embedding: Vector = embed( input=lambda: Document.content, provider="openai", model="text-embedding-ada-002" ) ``` By default, vectors are persisted as 16 bit floats for efficiency. However, Chalk also supports persisting vectors as 32-bit or 64-bit floats via the keywords "f32" and "f64", respectively. ``` from chalk.features import Vector @features class Document: # Defines a vector with 1536 dimensions that will be persisted with 32 bit precision embedding: Vector["fp32", 1536] ``` ### Custom serializers Finally, if you have an object that you want to serialize that isn't from dataclass, attrs, or pydantic, you can write a custom codec. This custom codec must target a type that can be serialized to a PyArrow data type, which is the underlying serialization format for all features. Consider the custom class below: ``` class CustomStruct: def __init__(self, foo: str, bar: int) -> None: self.foo = foo self.bar = bar def __eq__(self, other: object) -> bool: return ( isinstance(other, CustomStruct) and self.foo == other.bar and self.bar == other.bar ) def __hash__(self) -> int: return hash((self.foo, self.bar)) ``` Here, we use the custom class as a feature, and provide an encoder and decoder. The encoder takes an instance of the custom type and outputs a Python object, and the decoder takes output of the encoder and creates an instance of the custom type ``` @features class Book: custom_field: CustomStruct = feature( encoder=lambda x: dict(foo=x.foo, bar=x.bar), decoder=lambda x: CustomStruct(**x), ) ``` # Has One source: https://docs.chalk.ai/docs/has-one ## Define one-to-one relationships between feature classes. Has-one relationships link a feature to a single instance of another feature. ### Recommended Implementation The simplest way to specify a join for a has-one relationship is implicitly. In the example below, a User is linked to their Profile. ``` from chalk.features import features @features class Profile: id: str user_id: "User.id" email_age_years: float @features class User: id: str profile: Profile ``` With a has-one relationship established, you can reference features on Profile through User. For example: ``` user_email_age = User.profile.email_age_years ``` ### Explicit Join In the following snippet, the has-one join is explicitly defined. This is functionally equivalent to the recommended implementation: ``` from chalk.features import features, has_one, ... @features class Profile: id: str user_id: str email_age_years: float @features class User: id: str uid: str profile: Profile = has_one(lambda: Profile.user_id == User.uid) ``` The lambda solves forward references, letting you reference User before it is defined. ### Composite Join Keys You can also specify a composite join key for a has-one relationship. For example, if a User is linked to a Profile by org and email, you can define the join as follows: ``` from chalk.features import features, has_one from datetime import datetime @features class User: id: str email: str = _.alias + "-" + _.org + _.domain org_domain: str = _.org + _.domain org: str domain: str alias: str # join with composite key posts: DataFrame[Posts] = has_many(lambda: User.email == Post.email) # multi-feature join org_profile: Profile = has_one(lambda: (User.alias == Profile.email) & (User.org == Profile.org)) @features class Workspace: id: str # join with child-class's composite key users: DataFrame[Users] = has_many(lambda: Workspace.id == User.org_domain) ``` ### Back-references ### One-to-one You can also add a back-reference to User from Profile. However, you don't have to explicitly set the join on Profile. Instead, the join condition is assumed to be symmetric and copied over. To complete the one-to-one relationship from our example, add a User to the Profile class: Here you need to use quotes around User to use a forward reference. ### Optional relationships When a has-one relationship is specified, the default behavior is to treat the linked Feature as required. Following the example above, specifying a User without a Profile and querying for a User's profile or using the User.profile in a resolver raises an error. To define optional relationships, use the typing.Optional[...] keyword: Note, resolvers that take optional features as inputs need to handle the None case. This is covered in more detail in the resolver's section of the docs. ### Chained Has-One Joins You can chain has-one joins to traverse multiple relationships. For example, you could define the following features to represent a user's profile and preferences in an application. ``` from chalk.features import features, Primary @features class User: id: str email: str @features class Profile: id: Primary[User.id] username: str @features class Preferences: id: Primary[Profile.id] dark_mode: bool ``` ### Organizing Feature Definitions Across Files In large projects, it's common to split feature definitions across multiple Python modules. For unidirectional dependencies, this is straightforward. For example, if src/models/user.py imports src/models/profile.py, you can define the User and Profile features in separate files without issues. However, if you have circular dependencies, you may run into problems. Chalk supports this, but circular imports can arise when features reference each other across files. To avoid these issues, use the if TYPE_CHECKING block from the typing module and quote your forward references. Here's an example of how to do this cleanly: ``` # Imports `User` directly, because `src/models/user.py` # doesn't import `src/models/profile.py` from chalk.features import features from src.models.user import User @features class Profile: id: Primary[User.id] username: str ``` ``` from chalk.features import features from typing import TYPE_CHECKING if TYPE_CHECKING: # Imports `Profile` only when type checking # to avoid circular imports from src.models.profile import Profile @features class User: id: str # Profile must be quoted because it is imported # only when type checking profile: "Profile" ``` By quoting imports inside if TYPE_CHECKING, you avoid circular dependency errors while still maintaining type safety and feature linkage. If the relationship to Profile is optional, you can use typing.Optional or the | None syntax, but the entire annotation should be quoted: ``` from chalk.features import features from typing import TYPE_CHECKING if TYPE_CHECKING: from src.models.profile import Profile @features class User: id: str # All of `"Profile | None"` must be quoted, not just the `Profile` part profile: "Profile | None" ``` ### Querying for Has-One Relationships You can also query for a feature that is joined through a has-one relationship by referencing the root namespace. For example, to query for the Profile features associated with a User continuing from the example above, you can write: ``` from chalk.client import ChalkClient from src.features import User client = ChalkClient() client.query( input={User.id: "1"}, output=[User.profile.id, User.profile.email_age_years] ) ``` # Has Many source: https://docs.chalk.ai/docs/has-many ## Define one-to-many and many-to-many relationships between feature classes. Has-many relationships link a feature to many instances of another feature. ### Foreign Keys The recommended way to specify a join for a has-many relationship is implicitly. In the example below, a User is linked to potentially multiple Transfers. ``` from chalk.features import features, DataFrame @features class Transfer: id: str # note, the annotation must be a string reference because User is # defined after Transfer. user_id: "User.id" amount: float @features class User: id: str transfers: DataFrame[Transfer] ``` ### Explicit Join The following example, which explicitly sets the join, is equivalent to the above: ``` from chalk.features import has_many, DataFrame @features class Transfer: id: str user_id: str amount: float @features class User: id: str transfers: DataFrame[Transfer] = has_many(lambda: Transfer.user_id == User.id) ``` ### Composite Join Keys You can also specify multiple join keys for a has-many relationship. For example, say hospitals wants to compute aggregations over visit to each of their departments. We could write the following feature classes. ``` from chalk import _ from chalk.features import features, DataFrame, has_many import chalk.functions as F from datetime import datetime @features class HospitalVisit: id: str = _.user_id + _.hospital + _.department + F.cast(_.date, str) composite_key: str = _.hospital + "-" + _.department department: str hospital: str user_id: str date: datetime @features class HospitalDepartment: id: int name: str hospital_name: str composite_key_match: str = _.hospital_name + "-" + _.name # multi-feature join visits: DataFrame[HospitalVisit] = has_many( lambda: (HospitalDepartment.hospital_name == HospitalVisit.hospital) & (HospitalDepartment.name == HospitalVisit.department) ) # composite key join visits_with_composite_key: DataFrame[HospitalVisit] = has_many( lambda: HospitalDepartment.composite_key_match == HospitalVisit.composite_key ) ``` You can also join in Has-Many relationships for features that have primary composite keys. ``` from chalk.features import features, DataFrame, has_many @features class SoftwareEngineer: id: int = _.first_name + " " + _.last_name first_name: str last_name: str manager_id: str @features class Manager: id: int direct_reports: DataFrame[SoftwareEngineer] = has_many( lambda: Manager.id == SoftwareEngineer.manager_id ) ``` ### Aggregations on References Having established a has-many relationship, you can now reference the transfers for a user through the user namespace. The has_many feature returns a chalk.DataFrame, which supports many helpful aggregation operations: ``` # Number of transfers made by a user User.transfers.count() # Total amount of transfers made by the user User.transfers[Transfer.amount].sum() # Total amount of the transfers made by the user that were returned User.transfers[ Transfer.status == "returned", Transfer.amount ].sum() ``` To compute an aggregation over one or more time windows, see our docs on windowed aggregations. ### Back-references ### One-to-many In the reverse direction, a one-to-many relation is defined by a has_one relation (following the above example, a user has many transfers but a transfer has a single user). However, you don't have to explicitly set the join a second time. Instead, the join condition is assumed to be symmetric and copied over. To complete the one-to-many relationship from our example, add a User to the Transfer class: Here, you need to use quotes around User to use a forward reference. ### Many-to-many The recommended way to define a many-to-many relationship is through a joining feature class. For instance, to define a many-many relationship between Actors and Movies, you could write the following feature classes: ``` from chalk.features import features, DataFrame @features class Actor: id: int appearances: "DataFrame[MovieRole]" full_name: str # this will be used to demonstrate one of the # ways the joining feature can be populated movie_ids: list[int] @features class Movie: id: int title: str @features class MovieRole: id: str actor_id: Actor.id movie_id: Movie.id movie: Movie ``` Here you need to use quotes around DataFrame[MovieRole] to use a forward reference. This joining feature class can be populated by a SQL file resolver: ``` -- resolves: MovieRole -- source: PG SELECT id, actor_id, movie_id FROM movie_roles; ``` Alternatively, by a DataFrame-returning Python resolver (namespaced to one of the joined feature sets): ``` @online def get_actor_in_movie( a_id: Actor.id, movie_ids: Actor.movie_ids, ) -> Actor.appearances: return DataFrame([ MovieRole( id=f"{a_id}_{m_id}", actor_id=a_id, movie_id=m_id ) for m_id in movie_ids ]) ``` The joining feature class lets you: - query for movie features from the Actor namespace, and - use movie features in downstream Actor resolvers. For example, to get the titles for all the movies that an actor has appeared in, you can run the following query: ``` $ chalk query --in actor.id=1 --out actor.appearances.movie.title Results Name Hit? Value ─────────────────────────────────────────────────────────────────────────────── actor.appearances.movie.title ["The Bad Sleep Well","High and Low",...] ``` # Composite Keys source: https://docs.chalk.ai/docs/composite-keys ## Model composite identifiers and joins in Chalk, including filter pushdown, SQL resolver outputs, and materialized aggregations. A composite key identifies a row or entity using more than one source value, such as workspace_id and user_id. In Chalk, you can model that relationship in two ways: - join on the individual component features, or - compute a single scalar feature that contains the composite key and join on that feature. Both patterns are valid. The important rule is that Chalk joins and filters on features, not on the implied parts of a computed string. If WorkspaceMember.id is computed from workspace_id and user_id, Chalk treats WorkspaceMember.id as its own feature. Chalk does not automatically split that value back into workspace_id and user_id later. This matters for both correctness and performance. The query must have the features needed to make the join possible, and large offline queries must also have join keys that can be pushed down safely so the warehouse can prune the data it scans. ### Component Joins Use a component join when both sides naturally expose the same component columns. This usually gives warehouses the clearest filter predicates because each component can be pushed down independently. ``` from chalk import DataFrame from chalk.features import features, has_many @features class ActivityEvent: id: str workspace_id: str user_id: str amount: float @features class WorkspaceMember: id: str workspace_id: str user_id: str activity_events: DataFrame[ActivityEvent] = has_many( lambda: (WorkspaceMember.workspace_id == ActivityEvent.workspace_id) & (WorkspaceMember.user_id == ActivityEvent.user_id) ) ``` If an offline query starts from WorkspaceMember.workspace_id and WorkspaceMember.user_id, Chalk can push those inputs toward SQL resolvers as filters like: ``` where workspace_id in (...) and user_id in (...) ``` This is often the best shape when the underlying table is clustered, partitioned, or indexed by the component columns. ### Scalar Composite Keys Use a scalar composite key when the rest of the graph needs a single feature to represent the identity. This is common when you want the id of a feature class to be derived from multiple source columns. ``` from chalk import _ import chalk.functions as F from chalk.features import feature, features @features class WorkspaceMember: workspace_id: str user_id: int id: str = feature( expression=_.workspace_id + ":" + F.cast(_.user_id, str), primary=True, ) ``` The expression must be deterministic and must produce exactly the same value everywhere the key is used. When a query provides all of the input features used by the expression, Chalk can synthesize the primary key from those inputs. For example, if the query provides WorkspaceMember.workspace_id and WorkspaceMember.user_id, Chalk can compute WorkspaceMember.id. All inputs to the expression must be available. If the query provides workspace_id without user_id, Chalk cannot partially compute the key and will ask for the missing component. If a child feature class needs to join to this key, define the child-side foreign key as a feature too: ``` from chalk import _ import chalk.functions as F from chalk.features import feature, features, has_one WORKSPACE_MEMBER_ID = _.workspace_id + ":" + F.cast(_.user_id, str) @features class WorkspaceMember: workspace_id: str user_id: int id: str = feature(expression=WORKSPACE_MEMBER_ID, primary=True) @features class ActivityEvent: id: str workspace_id: str user_id: int workspace_member_id: "WorkspaceMember.id" = feature(expression=WORKSPACE_MEMBER_ID) workspace_member: WorkspaceMember = has_one( lambda: _.workspace_member_id == WorkspaceMember.id ) ``` The string annotation "WorkspaceMember.id" declares that workspace_member_id is a foreign key to WorkspaceMember.id. It does not compute the value. The value computation comes from feature(expression=...). ### SQL Resolver Outputs SQL resolvers must output the features that Chalk needs to join or filter on. This is the most common source of confusion with composite keys. Suppose your relationship joins on _.workspace_member_id == WorkspaceMember.id. If the resolver for ActivityEvent only emits workspace_id and user_id, the planner does not have a workspace_member_id column to attach the join or filter to. In an offline query, that can force Chalk to unload or scan far more of the table than expected. Even if the resolver SQL computes a composite value internally, Chalk can only push filters to features that are present in the resolver output and used by the relationship. For a scalar composite key join, emit the scalar key: ``` -- source: snowflake -- resolves: ActivityEvent select id, workspace_id, user_id, workspace_id || ':' || cast(user_id as string) as workspace_member_id, amount from activity_events ``` For a component join, emit the component columns: ``` -- source: snowflake -- resolves: ActivityEvent select id, workspace_id, user_id, amount from activity_events ``` If a query spine provides only workspace_member.id, but the resolver exposes only workspace_id and user_id, Chalk cannot infer the component filters from the scalar id. Either provide workspace_id and user_id directly as inputs, or emit the scalar composite key as a resolver output and join on that feature. There is one useful exception for root resolvers that return rows for a feature class whose primary key has an underscore expression. If the resolver returns every component needed by the primary-key expression, Chalk can synthesize the primary key from those columns. If the primary key does not have an underscore expression, the resolver must output the primary key itself. ### Materialized Aggregations For materialized aggregations over a has_many, Chalk keys the materialized values by the join features used by the relationship. When an aggregation is reached through a has_one prefix, the has_one join keys and the has_many local keys must overlap. This can matter in two-hop graphs: ``` from chalk import DataFrame, Windowed, _, windowed import chalk.functions as F from datetime import datetime from chalk.features import feature, features, has_many, has_one WORKSPACE_MEMBER_ID = _.workspace_id + ":" + F.cast(_.user_id, str) @features class ActivityEvent: id: str workspace_id: str user_id: int workspace_member_id: "WorkspaceMember.id" = feature(expression=WORKSPACE_MEMBER_ID) occurred_at: datetime amount: float @features class WorkspaceMember: workspace_id: str user_id: int id: str = feature(expression=WORKSPACE_MEMBER_ID, primary=True) activity_events: DataFrame[ActivityEvent] = has_many( lambda: _.workspace_member_id == WorkspaceMember.id ) @features class MemberRiskFeatures: id: str workspace_member_id: "WorkspaceMember.id" workspace_member: WorkspaceMember = has_one( lambda: _.workspace_member_id == WorkspaceMember.id ) event_amount: Windowed[float] = windowed( "30d", materialization={"bucket_duration": "1d"}, expression=_.workspace_member.activity_events[ _.amount, _.occurred_at > _.chalk_window, _.occurred_at <= _.chalk_now, ].sum(), ) ``` Here, both relationships use WorkspaceMember.id as the key. That lets the materialization planner line up the has_one prefix with the has_many aggregation. If the has_one used WorkspaceMember.id but the has_many joined on (workspace_id, user_id), the planner may reject the materialization because the join keys do not overlap. ### Best Practices Prefer a source-native stable identifier when one exists. Use a computed composite key only when the source system does not already provide a durable id. Use component joins when the warehouse can prune efficiently on the component columns. This is especially useful for offline queries over large tables. Use scalar composite keys when the entity identity needs to travel through the graph as one feature, or when a materialized two-hop aggregation needs overlapping join keys. Keep the composite key expression in one place. Define a shared expression constant and reuse it for the parent primary key and any child foreign keys. Choose an encoding that cannot collide. If you concatenate strings, use a delimiter that cannot appear in the component values, escape the components, or use a stable hash of a structured value. Make casts explicit so every resolver and expression produces the same string. Emit every join key from SQL resolvers. If a relationship joins on a scalar composite key, the resolver should output that scalar key. If a relationship joins on components, the resolver should output the components. For root resolvers, either output the primary key or make sure the primary key has an underscore expression and the resolver outputs all of the expression inputs. Treat persisted composite keys as stable cache keys. Persisted results can be read back by the computed key, but changing the key expression changes the identity of cached rows. Test with a small offline query and inspect the query plan or generated SQL. You should see filters on the columns used by the relationship, not a full-table unload followed by a join in Chalk. Check both that the query can run and that the plan is efficient: a valid composite-key model can still be too expensive if the relevant filters are not pushed down. ### Common Pitfalls Assuming Chalk can split a composite id. If an input spine contains values like workspace-7:123, Chalk does not know that workspace-7 is workspace_id and 123 is user_id. Provide the component inputs or model and emit the scalar key directly. Providing only some key components. If WorkspaceMember.id is computed from workspace_id and user_id, Chalk needs both components to synthesize the id. Supplying only one component is not enough. Using a foreign-key annotation as a computation. A type like workspace_member_id: "WorkspaceMember.id" declares the relationship target. It does not populate workspace_member_id. Use feature(expression=...), a Python resolver, a SQL resolver column, or uploaded data to provide the value. Expecting a resolver to omit an uncomputable primary key. A DataFrame or SQL resolver may omit a computed primary key only when Chalk has an underscore expression and the resolver outputs all of that expression's inputs. Otherwise, the resolver must output the primary key feature. Joining on a key that the resolver does not output. If the planner needs activity_event.workspace_member_id, the resolver must return a column for workspace_member_id. Returning only workspace_id and user_id is not equivalent when the relationship joins on workspace_member_id. Mixing scalar and component joins across a two-hop materialization. If the first hop joins on WorkspaceMember.id but the second hop joins on workspace_id and user_id, the materialization planner may not be able to prove that the keys line up. Use the same join feature on both hops, or model the relationship with component joins consistently. Duplicating the key formula by hand. If the parent key changes but the child key expression does not, joins silently stop matching. Reuse a shared expression or, for SQL resolvers, keep the SQL projection in lockstep with the Chalk expression. # Chalk DataFrame source: https://docs.chalk.ai/docs/dataframe ## Describe and fetch rows of features. A Chalk DataFrame is a 2-dimensional data structure similar to pandas.DataFrame, but with richer underlying optimizations. In a Chalk DataFrame, the column headers are instances of chalk.feature. The Chalk DataFrame can be used to describe rows of data and as a wrapper for returning that data. The Chalk DataFrame is parameterized by the feature values it contains. For example: ``` DataFrame[User.name, User.email] ``` The DataFrame type is commutative on its type parameters. This property allows you to write the type parameters of a DataFrame in any order: ``` DataFrame[User.name, User.email] == DataFrame[User.email, User.name] ``` Note: The Chalk DataFrame is used as a feature type for defining has-many relationships between feature classes, but the functions, filters, and aggregations described in this document are meant to be used in Python resolvers or notebooks--not in Chalk Expressions. For information about the functions and aggregations that you can use in Chalk Expressions, see here. ### Constructing a DataFrame You're likely to use the Chalk DataFrame primarily as a type, or to handle one as the result of a SQL query. However, you can also construct a DataFrame directly. ### From a dictionary Much like a pandas.DataFrame, a chalk.DataFrame can be constructed from a dictionary. The keys of the dictionary are the headers and the values are the rows under each header. The keys can be specified as either strings ("taco.price") or as their Python types (Taco.price). You can convert from a feature's Python definition to a string with a cast: str(Taco.price) == "taco.price". ``` chalk_df = DataFrame( { Taco.id: ["t_1", "t_2", "t_3"], "taco.price": [1, 2, 10], str(Taco.contains_meat): [True, True, False], } ) ``` In the above example, we use all forms of referencing a feature's name. ### From a list You can create the same chalk.DataFrame by passing a list of feature classes: ``` chalk_df = DataFrame([ Taco(id="t_1", price=1, contains_meat=True), Taco(id="t_2", price=2, contains_meat=True), Taco(id="t_3", price=10, contains_meat=False), ]) ``` ### From a pandas.DataFrame You can also convert to and from a pandas.DataFrame. ``` import pandas as pd DataFrame( pd.DataFrame({ "taco.id": ["t_1", "t_2", "t_3"], # Taco.id or "taco.id" are accepted "taco.price": [1, 2, 10], }) ) ``` The Chalk DataFrame can be converted to a pandas.DataFrame via the method .to_pandas(): ``` df = DataFrame(...) df.to_pandas() ``` ### From a SQL query Chalk's SQL integrations output the type DataFrame. For more information, see the SQL Integration section. ``` pg = PostgreSQLSource() @offline def fn() -> DataFrame[Login.ts, Login.user_id, Login.status]: return pg.query( Login( status=LoginHistorySQL.status, ts=LoginHistorySQL.created_at, user_id=LoginHistorySQL.user_id, ) ).all() ``` ### Aggregations | Method | Description | | ----------------- | ------------------------------------------------------------ | | DataFrame.count() | The number of rows in a `DataFrame` | | DataFrame.mean() | The average for each column in the `DataFrame`† | | DataFrame.sum() | The sum of each column in the `DataFrame`† | | DataFrame.max() | The maximum value for each column in the `DataFrame`† | | DataFrame.min() | The minimum value for each column in the `DataFrame`† | † There will be only one row in the DataFrame after this operation, one for each value. If the DataFrame contained only a single column to start, then the value returned is a scalar. ### Projections You can select columns out of a DataFrame from the set of columns already present to produce a new DataFrame scoped down to those columns. For example: ``` df: DataFrame[User.name, User.email, User.age] = ... projected = df[User.email, User.age] # type(projected) == DataFrame[User.email, User.age] ``` ### Filters In addition to selecting columns, you can filter the rows of a DataFrame. The example below restricts the rows of a DataFrame to only rows where the User is over 21 years old: ``` df: DataFrame[User.name, User.email, User.age] f = df[User.age >= 21] # type(f) == DataFrame[User.name, User.email, User.age] ``` Filtering a DataFrame keeps all the existing columns, but drops rows where the predicate is not met. The Chalk DataFrame supports the standard comparison functions (<, <=, >, >=, in, is, is not, ==, and !=) for features and their matching scalars values. You can compose these filters with the boolean operations and, or, and not. | Operation(s) | Example | Comment | | ------------ | ----------------------------------------------------------- | ------------------------- | | >, <, >=, <= | User.age > 21 | | | ==, != | User.age is None; User.age is not None | AST-only feature.† | | is, is not | User.age is None; User.age is not None | AST-only feature.† | | in, not in | User.age in {17, 18}; User.age not in [17, 18] | AST-only feature.† | | or, and | (User.state == "AL" and User.age >= 17) ; or User.age >= 18 | AST-only feature.† | | not | not (User.state == "AL" and User.age >= 17) | AST-only feature.† | † The marked operations must be provided as direct arguments. They cannot first be assigned to a variable, and later provided as an argument. For example, DataFrame[User.age in {17, 18}] is valid, but defining x = User.age in {17, 18} and then running DataFrame[x] is not valid. The reason for this is that Python does not allow overriding these operations. However, Chalk's aim is to allow developers to inspect and support natural Python syntax. To that end, Chalk parses the AST to override these operations, with the restriction that they are provided as arguments directly. ### Composing Filters To filter on multiple columns, separate your filters by a comma: ``` df[User.age > 21, User.email == "joe@chalk.ai"] ``` Alternatively, use the and keyword between filters: ``` df[User.age > 21 and User.email == "joe@chalk.ai"] ``` Similarly, you can perform or queries with Python's or keyword: ``` df[User.age > 21 or User.email == "joe@chalk.ai"] ``` All of these can be composed: ``` df[ User.age > 21 and ( User.email == "joe@chalk.ai" or User.full_name is None ) ] ``` ### Composing projections and filters Filtering may be combined with projection. For example, you can select all the values for User.email where the user is at least 21 years old as follows: ``` df: DataFrame[User.name, User.email, User.age] projected = df[User.email, User.age > 21] # type(projected) == DataFrame[User.email] ``` If you'd like to filter on a value and also return the value, you need to explicitly select it. You can amend the above example to include the User.age feature like so: ``` df: DataFrame[User.name, User.email, User.age] projected = df[User.email, User.age, User.age > 21] # type(projected) == DataFrame[User.email, User.age] ``` ### Performance (Vectorization, Laziness) Chalk's DataFrame is designed to support efficient operations across a variety of underlying data sources. DataFrame vectorizes scalar operations on data that is kept resident in memory. Chalk calls operations on data that is loaded into memory "strict execution". In coordination with Chalk's Execution Engine, DataFrame is capable of pushing down filtering, projection, and aggregation operations to underlying data sources. Chalk calls this style of execution "lazy execution". Suppose that we use a DataFrame to query a SQL source: ``` @online def get_return_count( transfers: User.transfers[Transfer.status == "returned", after(days_ago=60)] ) -> User.returned_transactions_last_60: return transfers.count() ``` This ultimately results in (approximately) the following SQL query being executed: ``` SELECT COUNT(*) from transfers WHERE status = 'returned' and transfers.ts > current_date - interval '60' day; ``` This "push down" mechanism in lazy computation helps make computationally expensive operations execute quickly, since Chalk only needs to load a single integer from the underlying data source, instead of potentially all Transfer rows. # Validation source: https://docs.chalk.ai/docs/validation ## Validate features and resolvers ### Feature Values Chalk can enforce requirements on your feature values. You can validate the values and length of many primitive types through the keyword arguments min, max, min_length, and max_length. To prevent 'invalid' features from being written to the offline or online store, set strict=True. In this case an error will be thrown when an invalid feature is observed. The validations keyword argument can be used when there are multiple validations, only some of which are strict. ``` from chalk import Validation from chalk.features import feature, features @features class Office: size_sqft: int = feature(min=0, max=100_000_000) street: str = feature(validations=[ Validation(min_length=1, max_length=256, strict=True), Validation(min_length=20, max_length=100) ]) ``` ### Feature validation requirements For some features, it is important to specify metadata such as owner and description. Chalk allows you to enforce metadata requirement easily. In addition, you can specify for each feature the severity at which to raise a missing metadata issue. In chalk.yml, the validation > feature > metadata section specifies settings for metadata validation. In the following example, users will be blocked from deploying features (chalk apply) with an unspecified owner. They are allowed to deploy features with missing description and tags. ``` project: Predict Q2 Spending validation: feature: metadata: - name: owner missing: error - name: description missing: warning - name: tags missing: info environments: default: runtimes: 'python310' requirements: requirements.txt ``` # Stream Resolvers source: https://docs.chalk.ai/docs/streams ## Define streaming resolvers with Chalk expressions. Stream resolvers consume messages from a streaming source — such as Kafka, Kinesis, or PubSub — and write Chalk features in real time. Resolvers are defined with statically-typed expressions that are compiled and executed as vectorized C++, enabling high-throughput processing. For source configuration (credentials, topic names, late-arrival policies), see Stream Sources. For local and end-to-end testing of resolvers, see Testing Stream Resolvers. ### Overview Stream resolvers are defined using the make_stream_resolver() function. The definition takes a message type and a mapping from Chalk features to expressions. In production, raw streaming messages are first converted into the given message type. Then, the output features of the resolver are computed based on the given expressions. ### Basic Example Let's start with a simple example that processes user data from a Kafka stream: ``` from chalk import Features, Primary, features from chalk.features import _ from chalk.features.resolver import make_stream_resolver from chalk import functions as F from pydantic import BaseModel import pyarrow as pa import datetime as dt # Define the Kafka stream source kafka_source = KafkaSource(name="user_updates") # Define the feature class @features(max_staleness="30d", etl_offline_to_online=True) class User: user_id: Primary[int] full_name: str email_address: str | None updated_at: dt.datetime | None # Define the message schema class ContactInfo(BaseModel): phone_number: str email_address: str | None class UserMessage(BaseModel): id: int # maps to user_id name: str updated_at: int # epoch microseconds contact_info: ContactInfo # Create a stream resolver users_streaming_resolver = make_stream_resolver( name="get_users_stream", # The resolver name source=kafka_source, # Your Kafka stream source message_type=UsersMessage, output_features={ User.user_id: _.id, User.full_name: _.name, User.email_address: _.contact_info.email_address, User.updated_at: F.from_unix_seconds(_.updated_at / 1_000_000), }, ) ``` Here, stream messages will come in as JSON strings whose structure matches the definition of the UserMessage model. The output features are defined as simple field accesses into the JSON struct. Note that nested access, such as _.contact_info.email_address, works as expected. Similar to feature definitions, scalar functions can be used to process the raw message fields. Here, we convert a raw integer timestamp in microseconds into a UTC datetime. Note that the expressions here are essentially the same as those used in feature definitions. However, the semantics are slightly different: The "root" here (), refers not to a namespace of features, but rather a table column whose datatype is based on the stream resolver's message_type. Thus, .amount does not mean "get me Chalk feature amount in the current namespace" but rather "access struct field amount on this column." In addition, since these expressions represent projections on single columns, dataframe operations such as .transactions[.amount, _.amount > 100].sum() don't apply here. ### Example Stream Messages Here's what an incoming message might look like: ``` { "id": 123, "name": "John Smith", "updated_at": 1700000000000000, "contact_info": { "phone_number": "555-555-5555", "email_address": "john.smith@gmail.com", } } ``` This message will get parsed into the following data: ``` User( id=123, full_name="John Smith", email_address="john.smith@gmail.com", updated_at=datetime.datetime(2023, 11, 14, 22, 13, 20, tzinfo=datetime.timezone.utc) ) ``` ### Supported message types The following message types are supported: - Pydantic BaseModels or python @dataclasses: Incoming string messages will be deserialized from JSON into structs. - Protocol buffer messages: The incoming binary messages will be deserialized into structs with the various fields. - str or bytes: The incoming messages can also be treated as simple strings or bytes. In this case, your feature expressions can't access struct fields on the message, but they can use scalar string/binary functions: ``` { Event.timestamp: _.timestamp # INVALID -- strings don't have an 'age' field Event.raw_data_cleaned: F.trim(F.lower(_)) # Here, `_` refers to the string message as whole. Event.timestamp: F.json_value(_, "$.metadata.timestamp") # This parses the input string as JSON and extracts a single field. } ``` Note: Not all pydantic or protobuf types are supported. The message type needs to be converted into a well-defined PyArrow schema. Types that are recursive and can be arbitrarily nested will fail to be parsed. Similarly, Pydantic types with type annotations such as Any or complex union types are also not supported. To process these types in a stream resolver, use a custom parse function to only extract specific fields with well-known datatypes. ``` class PhoneInfo(BaseModel): number: str area_code: str class EmailInfo(BaseModel): address: str class Parent(BaseModel): id: int parent: Parent | None class UsersMessage(BaseModel): id: int account_id: int | str # INVALID -- 'account_id' field's datatype is not well-defined card_id: int | None # Nullable types OK contact: PhoneInfo | EmailInfo # INVALID -- complex union types not supported parent: Parent | None # INVALID -- even if finitely nested in practice, the schema of this can't be determined ahead of time since it contains arbitrarily many sub-columns. ``` ### Custom Parse Functions By default, the message parsing logic is inferred automatically from the message_type. For Pydantic types or dataclasses, JSON parsing is used. For protobuf message classes, these are parsed from the protobuf's binary representation. However, it is also possible to provide custom parsing logic. In addition to handling custom message formats, this also allows the stream resolver to skip a subset of messages. For instance, one might have multiple message types on the same kafka topic, or perhaps a stream of events where only events of type "CREATE" are relevant. The make_stream_resolver() function has an optional parse argument that takes an expression. If provided, this expression will be used to convert raw message bytes into the expected message type. If the expression returns None, that message will be skipped. For example, consider the following protobuf definition: ``` message User { enum UserType { USER_TYPE_UNSPECIFIED = 0; USER_TYPE_GUEST = 1; USER_TYPE_MEMBER = 2; } message ContactInfo { optional string phone_number = 1; optional string email_address = 1; } uint32 id = 1; string name = 2; UserType user_type = 3; optional ContactInfo contact_info = 4; uint64 updated_at = 5; } ``` Let's say our system produces protobuf messages w/ an unusual format (a 6-byte prefix is added). We would like to strip the prefix before parsing and also ignore messages for users of type "GUEST". We can use a custom parse expression to do this: ``` from messages.user_pb2 import UserMessage # Import the generated protobuf classes from chalk import functions as F from chalk.features.resolver import make_stream_resolver # Strip first 6 bytes, then parse the protobuf message parsed_message = F.proto_deserialize(F.substr(_, 6, None), UserMessage) # Exclude any UserMessages whose user_type == GUEST parse_expression = F.if_then_else( parsed_message.user_type == UserMessage.UserType.GUEST, None, parsed_message ) users_streaming_resolver = make_stream_resolver( name="get_users_stream", source=kafka_source, message_type=UserMessage, parse=parse_expression, # Custom parse expression provided here output_features={ User.user_id: _.id, User.full_name: _.name, User.email_address: _.contact_info.email_address, User.updated_at: F.from_unix_seconds(_.updated_at / 1_000_000), }, ) ``` ### Fan-out: one message to many rows By default a stream resolver writes one feature row per message. When a single message carries a collection — a JSON array, or a repeated protobuf field — you may want one output row per element instead. To do this, write a parse expression that returns a list: Chalk explodes the list and runs output_features once per element. Two things change relative to the 1:1 case: - Set message_type to list[Element] — the list type is what tells the engine to explode. Element describes one output row's worth, and the parse expression produces the list of them. (A singular message_type=Element would not fan out.) - output_features expressions run per exploded element, so _ refers to one element at a time. An empty (or absent) list produces zero rows for that message — the natural way to express "this message contributes nothing." To skip a message entirely, a parse expression may also return None (see Custom Parse Functions). ### Fan out over a JSON array Use F.json_extract_array to pull the array out of the message, then F.array_transform to build one struct per element. Inside the transform you can reference both the current element and outer fields of the message, so per-element derived keys are straightforward: ``` class UserTagRow(BaseModel): user_id: str tag: str user_tags_resolver = make_stream_resolver( name="process_user_tags", source=kafka_source, message_type=list[UserTagRow], # list[Element] -> engine fans out one row per element parse=F.array_transform( # json_extract_array yields large_list; cast the whole array to # large_list so the lambda's element type unifies. F.cast(F.json_extract_array(F.bytes_to_string(_, "utf-8"), "$.tags"), pa.large_list(pa.large_string())), lambda tag: F.struct_pack( { # json_value also yields an arrow.json extension type — cast before string ops like `+`. "user_id": F.cast(F.json_value(F.bytes_to_string(_, "utf-8"), "$.user_id"), pa.large_string()), "tag": tag, } ), item_type=pa.large_string(), # type of each array element ), output_features={ # these run once per exploded element UserTag.id: _.user_id + ":" + _.tag, UserTag.user_id: _.user_id, UserTag.tag: _.tag, }, ) ``` A message with "tags": [] produces no rows. ### Fan out over a repeated protobuf field Suppose the message is an Order carrying a repeated items field: ``` message Order { string id = 1; repeated LineItem items = 2; } message LineItem { string id = 1; string sku = 2; } ``` Deserialize the message with F.proto_deserialize and transform its repeated items field. Each element is itself a struct, so array_transform needs an explicit item_type — pass the element's protobuf message class: ``` from messages.order_pb2 import Order, LineItem # your generated protobuf classes class LineItemRow(BaseModel): order_id: str item_id: str sku: str order_items_resolver = make_stream_resolver( name="process_order_items", source=kafka_source, message_type=list[LineItemRow], # list[Element] -> engine fans out per element parse=F.array_transform( F.proto_deserialize(_, Order).items, # repeated child field lambda item: F.struct_pack( { "order_id": F.proto_deserialize(_, Order).id, # parent field copied into each row "item_id": item.id, "sku": item.sku, } ), item_type=LineItem, # the element's protobuf message class ), output_features={ OrderLineItem.id: _.order_id + ":" + _.item_id, OrderLineItem.sku: _.sku, }, ) ``` For testing fan-out resolvers locally, see Testing a fan-out resolver. ### Filtering Multiplexed Topics with Message Headers A single Kafka topic often carries multiple message types — for example, a User event and a Transaction event distinguished by a header like X-Chalk-MessageType. The recommended way to route each type to the right resolver is the message_header_filters argument on make_stream_resolver. Each resolver declares the (header_name, header_value) pairs it wants to consume, and the streaming server drops any message whose Kafka headers don't match before it is deserialized. ``` users_streaming_resolver = make_stream_resolver( name="get_users_stream", source=kafka_source, message_type=UserMessage, output_features={ User.user_id: _.id, User.full_name: _.name, }, message_header_filters=[("X-Chalk-MessageType", b"User")], ) transactions_streaming_resolver = make_stream_resolver( name="get_transactions_stream", source=kafka_source, # same topic message_type=TransactionMessage, output_features={ Transaction.id: _.id, Transaction.amount: _.amount, }, message_header_filters=[("X-Chalk-MessageType", b"Transaction")], ) ``` A few things to keep in mind: - Header values are matched as raw bytes, not strings — use b"User", not "User". - Multiple entries in the list are combined with logical AND: a message must match every (name, value) pair to be processed. - The source must support headers. Kafka does; the validator will raise at deployment time if you attach filters to a source that doesn't. Prefer message_header_filters over inspecting headers inside a custom parse expression. Header filtering runs before deserialization, so messages destined for other resolvers are discarded cheaply without ever being parsed. If your filtering logic needs to look at the message body — not just headers — fall back to a parse expression that returns None for messages you want to skip (see Custom Parse Functions). You can also combine both: pre-filter on headers, then run additional body-level filtering inside parse. ### Performance Considerations Stream resolvers compile to vectorized C++, so a single CPU core can handle thousands of messages per second, and processing is parallelized across cores automatically. Expressions are validated at compile time, so schema and type mismatches surface before deployment rather than at runtime. Stream resolvers are ideal for: - High-volume data ingestion (>1000 messages/second) - Low-latency feature computation (<10ms) - Simple to moderate transformations - Stateless processing For complex business logic that requires external API calls or elaborate error handling, a regular Python @online resolver downstream of a stream resolver is usually a better fit. ### Writing to a Sink Stream resolvers can write computed features to downstream Kafka topics using the Sink parameter. This enables chaining of stream processing stages, where one resolver's output becomes another resolver's input. This pattern is particularly useful for: - Multi-stage processing pipelines: Break complex transformations into discrete stages - Feature routing: Direct specific features to specialized processing streams - Model inference workflows: Send features to GPU-accelerated resolvers for heavy computation The sink specifies which output features should be written to the destination stream. These features are computed by the resolver or by downstream resolvers that depend on the initial resolver's outputs. ### Example: Item Embedding Pipeline This example demonstrates a two-stage pipeline where items are first processed through a stream resolver, then routed to a GPU-accelerated resolver for embedding generation: ``` from chalk.streams import KafkaSource from chalk.features.resolver import Sink, make_stream_resolver from chalk.features import _ from chalk import online from pydantic import BaseModel from src.models import Item import numpy as np class ItemMessage(BaseModel): id: int name: str price: float description: str image_url: str new_items = KafkaSource( name="new_items", ) embedding_stream = KafkaSource( name="item_embeddings", ) make_stream_resolver( name="process_new_items", source=new_items, message_type=ItemMessage, output_features={ Item.id: _.id, Item.name: _.name, Item.price: _.price, Item.description: _.description, Item.image_url: _.image_url, }, sink=Sink( send_to=embedding_stream, output_features=[Item.id, Item.embedding], ), ) @online(resource_hint="gpu") def run_item_embedding_model( price: Item.price, description: Item.description, state: Item.state, image_url: Item.image_url, ) -> Item.embedding: """Mock generate item embedding using a pre-trained model.""" # In a real implementation, this function would load a pre-trained model # and generate an embedding based on the input features or make an API # call to an external service. return np.random.rand(128).tolist() ``` In this example: - The process_new_items resolver consumes messages from the new_items Kafka source and extracts basic item features - The sink configuration specifies that Item.id and Item.embedding should be written to the item_embeddings stream - When a message is processed, Chalk computes Item.embedding by calling the run_item_embedding_model resolver, which runs on a GPU-enabled instance - The computed features are then written to the item_embeddings topic This pattern allows the lightweight stream resolver to handle high-throughput message parsing while delegating compute-intensive embedding generation to specialized GPU resources. The sink acts as a bridge, automatically triggering the downstream computation and routing results to the appropriate stream. Messages can be encoded as either arrow IPC streams or as JSON. The fully qualified feature names are used as the column names in the output stream or as the JSON keys. ### Deduplication Stream resolvers can drop duplicate messages by passing a Deduplication policy to make_stream_resolver. For each incoming message, Chalk evaluates the on expression to compute a deduplication key. If that key has already been seen within the within retention window, the message is skipped — no features are written and no downstream computation is triggered. ``` from chalk.features.resolver import Deduplication, make_stream_resolver transactions_streaming_resolver = make_stream_resolver( name="get_transactions_stream", source=kafka_source, message_type=TransactionMessage, output_features={ Transaction.id: _.id, Transaction.amount: _.amount, Transaction.submitted: F.from_unix_seconds(_.submitted / 1_000_000), }, deduplication=Deduplication(on=_.id, within="24h"), ) ``` The on expression is a regular underscore expression over the parsed message, so the dedup key can be any field or computed value — for example _.id, a composite like F.concat(_.user_id, "-", _.event_id), or a hash. The within argument accepts a Chalk duration string ("24h", "7d") or a datetime.timedelta. ### Performance Overhead Deduplication is not free: every message incurs an extra round-trip to a key-value store (Redis Lightning or DynamoDB, depending on your online store configuration) to check for and record the dedup key. This adds latency per message and consumes storage proportional to the volume of unique keys seen within the retention window. Consider the tradeoff before enabling deduplication on very high-throughput streams, and keep the within window as short as your duplicate-tolerance requirements allow. Deduplication is currently supported only when the online store is Redis Lightning or DynamoDB. ### Skip Writing to the Offline Store If you already store a historical log of your stream messages in a data warehouse, you can disable stream persistence to the Chalk offline store. This can be done by setting the skip_offline parameter of make_stream_resolver: ``` users_streaming_resolver = make_stream_resolver( name="get_users_stream", source=kafka_source, message_type=UsersMessage, output_features={ User.user_id: _.id, User.full_name: _.name, User.email_address: _.contact_info.email_address, User.updated_at: F.from_unix_seconds(_.updated_at / 1_000_000), }, skip_offline=True, # don't write resolved events from stream to offline store ) ``` ### Environment-specific Stream Resolvers A stream resolver can be restricted to a specific environment through the environment parameter of make_stream_resolver: ``` @features(max_staleness="30d", etl_offline_to_online=True) class Transaction: transaction_id: Primary[int] submitted: dt.datetime amount: int class TransactionMessage(BaseModel): id: int # maps to transaction_id submitted: int # epoch microseconds amount: int # new resolver to experiment with new Transaction class experimental_streaming_resolver = make_stream_resolver( name="new_transaction_stream_resolver", source=kafka_source, message_type=UsersMessage, output_features={ Transaction.transaction_id: _.id, Transaction.timestamp: F.from_unix_seconds(_.submitted / 1_000_000), Transaction.amount: int }, environments=["dev"], # only deploy resolver in "dev" environment ) ``` ### Migrating from @stream Earlier versions of Chalk used the @stream decorator to define streaming resolvers in Python. Existing @stream resolvers continue to run, but new resolvers should use make_stream_resolver — it compiles to vectorized C++ and offers significantly better performance for high-throughput streams. Most @stream resolvers translate directly to make_stream_resolver by replacing the decorated function body with an output_features mapping: ``` from chalk.features.resolver import make_stream_resolver from chalk import stream, Features # Old: Python @stream decorator @stream(source=user_stream) def process_users(message: UsersMessage) -> Features[Users]: return Users( user_id=message.id, has_been_evicted_bool=message.qualification.has_been_evicted, monthly_income_min=message.qualification.monthly_income.min, # ... more processing ) # New: make_stream_resolver process_users_native = make_stream_resolver( name="process_users_native", message_type=UsersMessage, output_features={ Users.user_id: _.id, Users.has_been_evicted_bool: _.qualification.has_been_evicted, Users.monthly_income_min: _.qualification.monthly_income.min, # ... same mappings using underscore expressions }, source=user_stream, ) ``` Both resolvers produce identical results. # Testing Stream Resolvers source: https://docs.chalk.ai/docs/streaming-testing ## Locally verify stream resolver parsers and end-to-end scenarios. Chalk provides three tools for testing stream resolvers, each with a different scope. Pick the one that matches what you want to verify: | Tool | Scope | When to use | | -------------------------------------------------------------------------------------------- | ------------------------------------------ | --------------------------------------------------------------------------------------------- | | [`check_stream_parsing`](#local-parser-checks-with-check_stream_parsing) | Parser only, local | Fast iteration on the `parse` expression and `output_features` mapping | | [`check_stream_scenario`](#end-to-end-scenarios-with-check_stream_scenario) | Full pipeline against a branch, multi-step | Verify materialized windowed aggregates, or behavior that depends on prior online-store state | | [`test_streaming_resolver`](#testing-against-a-deployed-branch-with-test_streaming_resolver) | Full resolver against a deployed branch | Sanity-check a deployed resolver against real bytes; inspect the resulting Arrow table | ### Local parser checks with check_stream_parsing The parsing logic of a stream resolver can be tested locally via chalk.testing.check_stream_parsing. This verifies that your parse expression and output_features mappings produce the expected results, without deploying to a branch or running the full resolver pipeline. ``` import datetime import json import pyarrow as pa from pydantic import BaseModel from chalk import functions as F from chalk.features import _ from chalk.features.resolver import make_stream_resolver from chalk.testing import StreamMessage, check_stream_parsing class UserMessage(BaseModel): id: int name: str updated_at_us: int users_streaming_resolver = make_stream_resolver( name="get_users_streaming", source=kafka_source, message_type=UserMessage, parse=F.struct_pack( { "id": F.cast(F.json_value(F.bytes_to_string(_, "utf-8"), "$.id"), pa.int64()), "name": F.json_value(F.bytes_to_string(_, "utf-8"), "$.name"), "updated_at_us": F.cast( F.json_value(F.bytes_to_string(_, "utf-8"), "$.updated_at_us"), pa.int64() ), } ), output_features={ User.user_id: _.id, User.full_name: _.name, User.updated_at: F.from_unix_seconds(_.updated_at_us / 1_000_000), }, ) check_stream_parsing( users_streaming_resolver, [ StreamMessage( message=json.dumps( {"id": 123, "name": "John Smith", "updated_at_us": 1_700_000_000_000_000} ).encode(), parsed=User( user_id=123, full_name="John Smith", updated_at=datetime.datetime( 2023, 11, 14, 22, 13, 20, tzinfo=datetime.timezone.utc ), ), ), StreamMessage( message=json.dumps( {"id": 456, "name": "Alice Liddell", "updated_at_us": 1_710_000_000_000_000} ).encode(), parsed=User( user_id=456, full_name="Alice Liddell", updated_at=datetime.datetime( 2024, 3, 9, 16, 0, 0, tzinfo=datetime.timezone.utc ), ), ), ], ) ``` Each StreamMessage contains the raw message bytes and the expected parsed feature output(s). The parsed field controls what is checked: - parsed=Feature(...) — assert the message produces exactly one row matching these fields. - parsed=[Feature(...), Feature(...)] — for a fan-out resolver (list-returning parse), assert it produces exactly these rows, in emission order. - parsed=[] — assert the message produces zero rows (for example an empty array or a filtered message). Use parsed=[], not parsed=None, when you want to verify that a message was dropped. ### Testing a fan-out resolver For a resolver whose parse returns a list, pass a list of expected rows — one per emitted element. The empty case is asserted with parsed=[]. The user_tags_resolver below is the JSON-array fan-out resolver from the stream resolver docs: ``` check_stream_parsing( user_tags_resolver, [ # one message with two tags -> two rows StreamMessage( message=json.dumps({"user_id": "u1", "tags": ["red", "blue"]}).encode(), parsed=[ UserTag(id="u1:red", user_id="u1", tag="red"), UserTag(id="u1:blue", user_id="u1", tag="blue"), ], ), # tag-less message -> zero rows ([] asserts emptiness; None would skip the check) StreamMessage( message=json.dumps({"user_id": "u2", "tags": []}).encode(), parsed=[], ), ], ) ``` ### End-to-end scenarios with check_stream_scenario check_stream_parsing verifies the parser in isolation, but it does not exercise the rest of the streaming pipeline: feature uploads, materialized windowed aggregates, or any computation that depends on prior online-store state. To test those behaviors, use ChalkClient.check_stream_scenario, which runs the full pipeline against a branch. The method accepts a sequence of step objects that are processed left-to-right: - StreamMessage(message=..., stream_resolver=..., timestamp=...) — parses message locally using the resolver's parse expression (the same path as check_stream_parsing), then uploads the resulting features against branch with materialized-aggregate updates enabled. Set timestamp to control the feature time of the uploaded row (useful for out-of-order or windowed tests). - UploadFeatures({Feature: value_or_list, ...}) — primes the branch online store directly without going through a resolver. Each value can be a scalar (uploaded as one row) or a list (bulk upload, one row per element). All features in a single UploadFeatures step must agree on row count. This path does not refresh materialized aggregates. - FeatureAssertion(input=..., output=...) — runs an online query whose inputs are the set fields of input and whose requested outputs are the set fields of output, then compares each returned value against the expected value. A mismatch raises AssertionError. For Windowed[...] features, pass a {window: value} dict on either side, e.g. transaction_count={"1d": 1, "30d": 7}, and each bucket is checked independently. check_stream_scenario parses each StreamMessage locally via the same chalkdf-backed path as check_stream_parsing, so the chalkdf package must be installed in the test environment. Uploads and assertions run against a Chalk branch, so the target environment must have a branch online store configured. Without it, the uploads produced by each StreamMessage and UploadFeatures step have nowhere to land and the subsequent FeatureAssertion queries will not see them. The example below seeds three merchant rows, then drives a transaction stream and a user-creation stream, and finally asserts that the windowed transaction_count aggregate reflects the single transaction processed: ``` from datetime import datetime from chalk.client import ChalkClient from chalk.testing import FeatureAssertion, StreamMessage, UploadFeatures ChalkClient().check_stream_scenario( UploadFeatures({ Merchant.id: [501, 502, 503], Merchant.category: ["grocery", "restaurant", "fuel"], }), StreamMessage( timestamp=datetime(2026, 1, 1, 12, 20), message=b'{"id": 1, "user_id": 1, "merchant_id": 501, "amount": 10.0}', stream_resolver=transaction_stream, ), StreamMessage( timestamp=datetime(2026, 1, 1, 12, 20), message=b'{"id": 1, "name": "Bartleby"}', stream_resolver=create_user, ), FeatureAssertion( input=User(id=1), output=User(name="Bartleby", transaction_count={"1d": 1}), ), branch="my-branch", ) ``` For the common case where you just want to prime the online store before the first stream message, pass seed_online_store= instead of writing an explicit UploadFeatures step. It is equivalent to prepending one UploadFeatures step at the head of the scenario: ``` ChalkClient().check_stream_scenario( StreamMessage( message=b'{"user_id": 1, "amount": 10.0}', stream_resolver=transaction_stream, timestamp=datetime(2026, 1, 1, 12, 20), ), FeatureAssertion( input=User(id=1), output=User(transaction_count={"1d": 1}), ), seed_online_store={User.id: [1, 2, 3], User.name: ["Alice", "Bob", "Carol"]}, branch="my-branch", ) ``` Other useful parameters: - show_table=True — print the Chalk Feature Value Check Table for every FeatureAssertion step, not only on failure. On a mismatch the table is always printed, regardless of this flag. - float_rel_tolerance / float_abs_tolerance — tolerances applied to float comparisons inside FeatureAssertion checks (defaults 1e-6 and 1e-12). Values pass if either tolerance is satisfied. - environment — pin the scenario to a specific Chalk environment; otherwise the client's default is used. ### Testing against a deployed branch with test_streaming_resolver To test against a deployed branch, use ChalkClient.test_streaming_resolver: ``` from chalk.client import ChalkClient sample_messages = [ '{"id": 123, "name": "John Smith"}', '{"id": 456, "name": "Alice Liddell"}', ] client = ChalkClient() # 1. Test messages against production resolver: result = client.test_streaming_resolver(resolver="get_users_streaming", message_bodies=sample_messages) # 'test_streaming_resolver' returns a signed URL to a parquet file in cloud storage -- `result.features` downloads this and extracts the table. print(result.features) # 2. Test messages against a stream resolver on a branch: result = client.test_streaming_resolver(resolver="get_users_streaming", message_bodies=sample_messages, branch="new_users_stream_logic") print(result.features) # 3. Create a new resolver and test it: new_users_resolver = make_stream_resolver( source=..., name="get_users_streaming_new", message_type=..., output_features={...}, ) result = client.test_streaming_resolver(resolver=new_users_resolver, message_bodies=sample_messages) print(result.features) ``` # Resolvers Overview source: https://docs.chalk.ai/docs/resolver-overview ## Create resolvers to compute feature values. While feature classes define what your data looks like (e.g. the types of your features, their relationships, and their properties), resolvers define how your feature values are computed. If you haven't read the section on feature classes, you are strongly encouraged to do so. Understanding feature classes will help you understand resolvers. Resolvers have four main components: inputs, outputs, code, and run conditions. The first three of these are rather straightforward. The inputs to a resolver are the features it uses to compute its output. The outputs of a resolver are the features that it computes. The code of the resolver is the logic that transforms your input features into your output features. The last component, run conditions, is a little more nuanced. At a high level, Chalk is optimized for two different access patterns: online and offline. In online access, information about a single entity is being requested and features must be computed quickly and cached. In offline access, a large chunk of historical feature data is being requested (you can think of this as creating a dataset for model training or running an analytic query). Resolvers are always specified as either online or offline. We go into more detail about what this means in the next section where we talk about run conditions, but for now it is important to note that: 1). every resolver is either online or offline, and 2). different resolvers run depending on how you are requesting data from Chalk. ### Writing Resolvers Chalk allows you to write resolvers in three different ways: - as SQL queries, with comments specifying data sources and output features. - as python functions, with type annotations specifying the input and output features. - as an expression, inline in your feature definitions. Below is an example of how to write the three kinds of resolvers in Chalk for a User feature class. In this example, we assume that a Postgres data source called pg_users has been configured and connected to Chalk. First, let's define our User feature class: ``` from chalk.features import features @features class User: id: str name: str age: int hobbies: list[str] is_runner: bool # computed feature athleticism_score: float # computed feature ``` Now we define our resolvers: We resolve the id, name, age, and hobbies features for our users from the users table of our pg_users data source. When reading data from your data sources, you should use a SQL resolver: ``` -- The features given to us by the user. -- resolves: user -- type: online -- source: pg_users select id, name, age, hobbies from users; ``` To compute the is_runner feature for users, we simply check whether running is in the list of hobbies, which we can do with an expression. Chalk expressions utilize low-latency C++ for efficient execution. Chalk expressions can use any of the chalk.functions to express a variety of computations. ``` import chalk.functions as F from chalk.features import features, _ @features class User: id: str name: str age: int hobbies: list[str] is_runner: bool = F.contains(_.hobbies, "running") athleticism_score: float # computed feature ``` We now use a python resolver to compute the athleticism_score feature for our users. Python resolvers are useful for integrating existing libraries, and executing more complex logic, such as API calls. ``` from chalk import online from src.user import User from utils.athleticism import is_athletic @online def get_is_runner( hobbies: User.hobbies ) -> User.athleticism_score: num_athletic_hobbies = sum([is_athletic(hobby) for hobby in hobbies]) return num_athletic_hobbies / len(hobbies) ``` ### When Do Resolvers Run A question that often crops up at this point is, "I have defined a resolver, but how do I make it run?" Just like features, the idea behind resolvers is that they are declarative. Resolvers don't run when you define them, and they don't run exhaustively over your data. They run only in response to a request for the features that they are responsible for computing. In practice, this means that they run in response to "queries". Resolvers can also be run on a schedule or triggered. Either approach causes them to 1). execute, in batch-job fashion, across all data they find in your upstream data sources, and 2). write computed features to your offline store. A query is just a request for data. Chalk takes a query and determines which resolvers it needs to run, optimizes the execution order, and provides you with a response. For now, we'll hold off on discussing queries further, but feel free to skip ahead and get a sense of what Chalk queries are and how they work before diving deeper into resolvers. # SQL Resolvers source: https://docs.chalk.ai/docs/sql-resolvers ## Define feature resolvers as a SQL query. SQL resolvers enable you to load and transform data from your data sources as feature values. ### Defining SQL Resolvers There are two ways to define SQL resolvers. The first is to write a SQL query in a file with the .chalk.sql extension, and the second is to use the make_sql_file_resolver function, which will generate a SQL file resolver based on the function arguments at build time. ### SQL File Resolvers To define a SQL resolver in a file, create a file with the .chalk.sql extension, add comments to specify the data source the SQL query will run on and the feature class whose features it will resolve, and write the SQL query. ``` -- source: pg_users -- resolves: User select id, name, age, hobbies from users ``` The output column names in the SQL query must match the feature names in the feature class specified in the resolves comment. The data source specified in the source comment must also match the name of the data source configured in the Chalk dashboard. Each SQL file resolver must only resolve features within the same feature class, but a resolver does not have to resolve all features within the feature class. ### Generating SQL File Resolvers To generate SQL file resolvers at build time, use the make_sql_file_resolver function. This function enables you to define one or many SQL file resolvers at a time that may have slight variations, for example querying a different data source in a different environment. The following code would generate an equivalent SQL file resolver to the one above: ``` from chalk import make_sql_file_resolver make_sql_file_resolver( name="get_user", source="pg_users", resolves="User", query="select id, name, age, hobbies from users" ) ``` ### SQL Resolver Configurations As with Python resolvers, SQL resolvers can be configured through many optional parameters. For generating SQL file resolvers, these parameters are passed as arguments to the make_sql_file_resolver function. For SQL file resolvers, these parameters are specified in the comments at the top of the file. ``` -- These parameters are required in all SQL file resolvers -- source: pg -- resolves: User -- These parameters are optional in all SQL file resolvers -- type: online <-- default is "online," but can also be "offline" or "streaming" -- incremental: <-- read more about incremental resolvers here: https://docs.chalk.ai/docs/sql#incremental-queries -- mode: row -- lookback_period: 60m -- incremental_column: updated_at -- count: 1 <-- the number of rows the resolver should return. This can be set to 1, 'one', 'one_or_none', or 'all' -- timeout: 5m <-- the maximum time the resolver should run before timing out -- cron: 0 0 * * * <-- a cron expression to schedule the resolver to run at a specific time -- owner: 'helly@lumonindustries.com' <-- the email of the owner of the resolver. Alerts can be routed to owners. -- tags: ['user', 'profile'] <-- tags to help organize resolvers in the Chalk dashboard -- environment: 'production' <-- the environment in which the resolver should run -- total: True <-- whether this resolver returns all ID's of a given namespace. select id, name, age, hobbies, updated_at from users ``` For the full list of configurations, see here. ### Testing To test your SQL resolvers, we recommend setting up integration tests or iterating on a branch. # Expressions source: https://docs.chalk.ai/docs/expression ## Using Chalk expressions to define features ### Overview Chalk Expressions let you define features declaratively, using symbolic computation over your data. While you write expressions in idiomatic Python, they are compiled and executed as vectorized C++, enabling low-latency computation at serve time and high-throughput processing at train time. Expressions support a wide range of operations, including arithmetic, filtering, aggregations, and built-in functions like . For example, in a Transaction feature class, we can compute the subtotal of a transaction as the difference between its total and sales tax: ``` from chalk.features import features, Primary from chalk import _ @features class Transaction: id: int total: float sales_tax: float subtotal: float = _.total - _.sales_tax ``` The _ symbol refers to the current scope (here, the feature class Transaction) and is used to reference other fields on the same instance. Expressions like _.total - _.sales_tax are compiled into native execution plans that run efficiently in production. In addition to referencing fields on the same object, you can traverse relationships. If each Transaction is associated with a User, for example, you can compute a string similarity between the user’s name and the transaction memo: ``` from chalk.features import _, features + import chalk.functions as F @features class Transaction: ... amount: float memo: str user: "User" + name_match_score: float = F.jaccard_similarity( + _.user.name, _.memo + ) ``` Here, _.user.name follows the foreign key relationship from Transaction to User. The function F.jaccard_similarity is one of many built-in Chalk functions that operate on symbolic expressions. Expressions can also perform aggregations over related records. In a User feature class, we can compute aggregates like the number of large transactions or the total amount spent: ``` from chalk import _ from chalk.features import DataFrame, features @features class User: id: int name: str transactions: DataFrame[Transaction] # Total spend is nullable because the sum of an empty DataFrame is null total_spend: float | None = _.transactions[_.total].sum() # The count is never null, because an empty DataFrame has count 0 num_large_txns: int = _.transactions[_.total > 1000].count() ``` In this context, _ refers to the User instance when referring to _.transactions. But when you apply a filter, like _.transactions[_.total > 1000], the expression inside the brackets is evaluated in the context of each individual Transaction. That means _.total refers to the total field on each Transaction, not on the User. This scoped evaluation makes it easy to filter, project, and aggregate over related data. All expressions are statically analyzed, optimized to eliminate redundant computation, and executed as high-performance C++ at runtime. ### Scalar Functions Chalk expressions support a wide range of built-in functions for manipulating data, performing calculations, and transforming features. These functions can be used in expressions to operate on feature values, DataFrames, and other data types. ``` - from chalk import features, online + from chalk.features import _, features @features class Transaction: id: int total: float sales_tax: float + subtotal: float = _.total - _.sales_tax - @online - def get_subtotal(total: Transaction.total, sales_tax: Transaction.sales_tax) -> Transaction.subtotal: - return total - sales_tax ``` ### Infix Operators Chalk expressions support a variety of infix operators for arithmetic, conditions, and boolean logic. | Operator | Description | Example | | -------- | --------------------- | ----------------------------- | | `+` | Addition | `_.total + _.sales_tax` | | `-` | Subtraction | `_.total - _.sales_tax` | | `*` | Multiplication | `_.quantity * _.price` | | `/` | Division | `_.total / _.quantity` | | `>` | Greater than | `_.total > 1000` | | `>=` | Greater than or equal | `_.total >= 1000` | | `<` | Less than | `_.total < 1000` | | `<=` | Less than or equal | `_.total <= 1000` | | `==` | Equal | `_.status == "completed"` | | `!=` | Not equal | `_.status != "completed"` | | `&` | Boolean and | `_.is_active & _.is_verified` | | \| | Boolean or | _.is_active \| _.is_verified | | `~` | Boolean not | `~_.is_active` | Python does not allow these operators to be overridden, so they will not work with Chalk's expressions. Instead, use the infix operators &, |, and ~ for boolean logic, and use == and != for equality comparisons. ### Builtin Functions The chalk.functions module exposes several helpful functions that can be used in combination with expression references to transform features. These functions are meant to be used in expressions and are not available as standalone functions. To view all available functions, see our SDK docs. ### Structs Expressions can be used to access nested attributes from other features in a feature class, whether these other features are struct dataclasses, features, or DataFrames. ``` import chalk.functions as F from chalk.features import features from dataclasses import dataclass @dataclass class LatLon: lat: float | None lon: int | None @features class User: id: int home: LatLon work: LatLon commute_distance: float = F.haversine( lat1=_.home.lat, lon1=_.home.lon, lat2=_.work.lat, lon2=_.work.lon, ) ``` ### Custom Functions You can create custom functions to encapsulate complex logic or reusable computations in your expressions. For example, if you wanted to apply consistent windows across many features, you could define a custom function like this: Use helper functions to create expressions ``` from chalk import _ from chalk.features import features, DataFrame def count_where(*filters): return _.transactions[_.created_at > _.chalk_window, *filters].count() @features class User: id: int transactions: DataFrame[Transaction] num_large_transactions: int = count_where(_.total > 1000) num_small_transactions: int = count_where(_.total < 100) ``` Expressions are not supported in Python resolvers, so you cannot use Chalk functions like F.jaccard_similarity in a Python resolver. Don't use expressions in Python resolvers ``` @features class User: id: int name: str email: str name_email_match_score: float @online def get_score( name: User.name, email: User.email, ) -> User.name_email_match_score: # Don't do this!! Expressions don't run in Python resolvers return F.jaccard_similarity(name, email) ``` Instead, use expressions to define the feature directly in the feature class: Use expressions in feature classes ``` @features class User: id: int name: str email: str name_email_match_score: float = F.jaccard_similarity( _.name, _.email ) ``` ### DataFrame Functions ### Conditions and filters DataFrame features can be filtered with expressions. Extending our Transaction example, we can create a User feature class with a has-many relationship to Transaction. Then, we can define a feature representing the number of large purchases by referencing the existing User.transactions feature: ``` from chalk.features import _, features, DataFrame @features class Transaction: id: int + user_id: "User.id" total: float sales_tax: float subtotal: float = _.total - _.sales_tax + @features + class User: + id: int + # implicit has-many relationship with Transaction due to `user_id` above + transactions: DataFrame[Transaction] + num_large_transactions: int = _.transactions[_.total > 1000].count() ``` The object referenced by _ changes depending on its current scope. In this code, the _ in _.transactions references the User object. Within the DataFrame filter, the _ in _.total references each Transaction object as each one is evaluated. The count aggregation is covered in the next section. ### Projections and aggregations DataFrame features support projection with expressions, which produce a new DataFrame scoped down to the referenced columns. DataFrames can be aggregated after eligible columns are selected. With our Transaction example, we already saw a count aggregation for counting the number of large transactions. We can add another aggregation for computing the user's total spend: ``` from chalk.features import _, features, DataFrame @features class Transaction: id: int user_id: "User.id" sales_tax: float subtotal: float total: float = _.subtotal + _.sales_tax @features class User: id: int transactions: DataFrame[Transaction] num_large_transactions: int = _.transactions[_.total > 1000].count() + total_spend: float = _.transactions[_.total].sum() ``` To compute User.total_spend, we needed to create a projection of the User.transactions DataFrame limited to only the Transaction.total column so that the sum aggregation could work. In contrast, no projection was needed for the num_large_transactions aggregation because count works on DataFrames with any number of columns. For computing low-latency aggregations over high volumes of data, Chalk also offers materialized windowed aggregations that uses materialization of buckets of data to compute large aggregations efficiently. ### Aggregation functions Aggregation functions have varying behavior when handling None values and empty DataFrames. If an aggregation function says None values are skipped in the table below, it will consider a DataFrame with only None values as empty. | Function | `None` values | Empty DataFrame | Notes | | ----------------------- | ------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `sum` | Skipped | Returns `0` | | | `min` | Skipped | Returns `None` | | | `max` | Skipped | Returns `None` | | | `mean` | Skipped | Returns `None` | Feature type must be `float` or `float \| None`. `None` values are skipped, meaning they are not included in the mean calculation. | | `count` | Included | Returns `0` | | | `any` | Skipped | Returns `False` | | | `all` | Skipped | Returns `True` | | | `std` | Skipped | See notes | Standard deviation. Requires at least 2 values. For DataFrames with less than 2 values, returns `None`. ; Aliases: `stddev`, `stddev_sample`, `std_sample`. | | `var` | Skipped | See notes | Variance. Same requirements as `std`. ; Alias: `var_sample`. | | `approx_count_distinct` | Skipped | Returns `0` | | | `approx_percentile` | Skipped | Returns `None` | Takes one argument, `quantile`, expected to be a float in the range `[0, 1]`. ; Example: `approx_percentile(0.75)` returns a value approximately equal to the 75th percentile of the not-`None` values in the DataFrame. | | `approx_top_k` | Skipped | Returns `None` | Takes one argument, `k`, expected to be a positive integer, as a keyword argument. ; Example: `approx_top_k(k=25)` returns the 25 or fewer approximately-most common values. | For aggregations that can return None, either mark the feature as optional (for example, by setting the feature type to float | None) or use coalesce to fall back to a default value. ### Run conditions and Execution To specify run conditions such as environment, tags, and versions for a feature that is resolved through an expression, you can use the feature function and pass in the expression as an argument. ``` from chalk.features import Primary, features, feature @features class User: id: int purchases: DataFrame[Purchase] # Uses a default value of 0 when one cannot be computed. num_purchases: int = feature( expression=_.purchases.count(), default=0, environment=["staging", "dev"], tags=["fraud", "credit"], version=1, ) ``` You may also want to define expressions that are only meant to be used for the offline pathway, say if you would like to compute values or aggregations dependent on offline resolvers. To do so, you can specify a distinct offline_expression as an offline resolver for your feature. ``` from chalk.features import features, feature @features class Merchant: id: int transactions: DataFrame[Transaction] merchant_risk_score: float = feature( expression=_.transactions[_.status=="approved"].count() / _.transactions.count(), offline_expression=(0.50*_.transactions[_.is_chargeback==True].count()+0.25*_.transactions[_.is_refund==True].count()+0.25*_.transactions[_.status=="approved"].count())/_.transactions.count() ) ``` ### Testing To test your expressions, we recommend setting up integration tests or iterating on a branch. ### Dynamic Expressions In some cases, you may want to build expressions dynamically. For example, if you have a rules engine that generates expressions and stores them in a database, you can load those expressions at runtime and ask Chalk to compute their values. Consider this User and Transaction model, which would be checked in to the code: ``` from chalk.features import features, DataFrame @features class Transaction: id: int user_id: "User.id" user: "User" total: float sales_tax: float @features class User: id: int transactions: DataFrame[Transaction] name: str email: str ``` Using the Golang or Python SDKs, you can compute both scalar and aggregate features on demand. For example, in Golang, if we wanted to compare the lowercased name and email by their Jaccard similarity, we could build the expression dynamically and ask Chalk to compute it: ``` package main import ( "testing" "github.com/zeebo/assert" "github.com/chalk-ai/chalk-go/expr" "github.com/chalk-ai/chalk-go" ) func TestQueryingExpressions(t *testing.T) { // Picks up the ambient credentials from the `chalk login` run on the CLI. client, err := chalk.NewGRPCClient(t.Context()) assert.NoError(t, err) result, err := client.OnlineQueryBulk( t.Context(), chalk.OnlineQueryParams{}. WithInput("user.id", []int{1}). WithOutputs("user.name"). WithOutputs("user.email"). WithOutputExprs( expr.FunctionCall( "jaccard_similarity", expr.FunctionCall("lower", expr.Col("name")), expr.FunctionCall("lower", expr.Col("email")), ). As("user.name_email_sim"), ), ) assert.NoError(t, err) row, err := result.GetRow(0) assert.NoError(t, err) for feature, value := range row.Features { t.Logf("Feature: %s, Value: %+v", feature, value.Value) } } ``` This program produces output like this: ``` === RUN TestChalkClient chalk_test.go:46: Feature: user.name, Value: Nicole Mann chalk_test.go:46: Feature: user.email, Value: nicoleam@nasa.gov chalk_test.go:46: Feature: user.name_email_sim, Value: 0.5714285714285714 --- PASS: TestChalkClient (0.43s) ``` The feature user.name_email_sim was computed using the expression jaccard_similarity(lower(name), lower(email)), which was built dynamically using the expr package. For a complete list of functions, see our SDK docs. All the functions available in expressions are also available in the expr package. You can also compute aggregations dynamically. Using the Golang SDK, we can compute the number of large transactions for a user: ``` result, err := client.OnlineQueryBulk( t.Context(), chalk.OnlineQueryParams{}. WithInput("user.id", []int{1}). WithOutputExprs( expr.DataFrame("transactions"). Filter(expr.Col("amount").Gt(expr.Float(0.))). Agg("count"). As("user.positive_transaction_count"), ), ) ``` For this program, the output would look like this: ``` === RUN TestChalkClient chalk_test.go:42: Feature: user.positive_transaction_count, Value: 33 --- PASS: TestChalkClient (0.45s) ``` In a Python expression, the above SDK call is equivalent to: ``` User.positive_transaction_count = _.transactions[_.amount > 0].count() ``` Both the scalar functions and aggregations can be computed in Python SDK. See this guide for more details. # Python Resolvers source: https://docs.chalk.ai/docs/python-resolvers ## Specify dependencies and results of feature resolvers. Python resolvers are Python functions marked with either an @online or @offline decorator, with Python type annotations declaring the dependencies on other features, as well as the output feature(s). ### Inputs ### Scalar dependencies To depend on features from a feature class, label your resolver arguments with your features as the types. You can then use those arguments in the body of your resolver to compute your output features. If you're running our editor plugin, your editor will see the type of each variable as the type of the underlying scalar. ``` from chalk.features import features, online @features class User: id: str email: str email_domain: str @online def get_domain(email: User.email) -> User.email_domain: # type(email) == str return email.split('@')[1].lower() ``` You can require multiple features in a resolver. However, all feature dependencies in a single resolver need to originate in the same root namespace: Requiring features from the same root namespace ``` @online def get_name_sim(a: User.email, b: User.name) -> User.email_name_match: return len(set(a) & set(b)) / len(set(a) | set(b)) ``` Here, we incorrectly request features from the root namespaces of Transfer and User: Requiring features from different root namespaces ``` @online def txn_email_sim( email: User.email, memo: Transfer.memo, ) -> Transfer.email_in_memo: return email in memo ``` If you want to require features from namespaces, you can use has-one or has-many relationships. Requiring features from different root namespaces using a relationship ``` @online def txn_email_sim( email: Transfer.user.email, memo: Transfer.memo, ) -> Transfer.email_in_memo: return email in memo ``` ### Has-one dependencies ### Scalar has-one You can also require features joined to a feature class through has-one relationships. For example, if users in your system have bank accounts, and you wanted to compare the name on the user's bank account to the user's name, you could require the user's name and the account's title through the user: ``` @online def name_sim(title: User.account.title, name: User.full_name) -> User.name_sim: return len(set(title) & set(name)) / len(set(title) | set(name)) ``` You can also require all scalars on the user: ``` from chalk import online, Now @online def get_txn_name_match( memo: Transaction.memo, user: Transaction.user, ) -> User.profile.age: return ( len(set(memo) & set(user.full_name)) / len(set(memo) | set(user.full_name) ) ``` Chalk will materialize all scalar features on the user before calling this function. If you want to pull only a few features of the user, require each directly: ``` @online def get_txn_name_match( memo: Transaction.memo, full_name: Transaction.user.full_name, ) -> User.profile.age: return ( len(set(memo) & set(full_name)) / len(set(memo) | set(full_name) ) ``` ### Optional has-one Has-one relationships can also be declared as optional. You may also require feature through optional relationships, but the types for all of those optional features will become optional. Consider the below example: ``` from chalk.features import features, online @features class Account: id: int user_id: "User.id" balance: float # Non-optional balance @features class User: id: int account: Account | None has_high_balance: bool @online def has_high_bal(balance: User.account.balance) -> User.has_high_balance: # Balance will be "float | None", because User.account is optional if balance is None: return False return balance > 1000 ``` The resolver in this example receives an optional float, even though balance is not an optional field on Account. The optional is added because the user may not have an account, in which case the resolver will receive None for the balance. ### Nested has-one You can also traverse nested has-one relationships in the same manner as requiring a single has-one. Consider a schema where users have a feature class of profile information, and the user's profile has an identity feature class, which in turn has the age of the user's email. You can require the email age feature as below: ``` @online def email_score( email_age: User.profile.identity.email_age ) -> User.email_score: return ( 1 if email_age < 30 else 0.5 if email_age < 60 else 0 ) ``` However, you cannot access nested relationships without explicit asking for them. Accessing a transitive relationship from a dependency. ``` @online def fn(acct: User.account) -> ...: acct.balance # Ok acct.institution.name # Error! ``` Instead, you can require the nested relationship directly and access any of its scalar features. Directly requiring the transitive relationship. ``` @online def fn(ins: User.account.institution, acct: User.account) -> ...: acct.balance # Ok ins.name # Ok ``` The semantics of optional has-one dependencies carry over to nested has-one dependencies. If you traverse an optional relationship, then all downstream attributes will become optional. ### Has-many dependencies ### Scalar has-many You can also require has-many relationships as inputs to your resolver: ``` @online def txn_count(transfers: User.transfers) -> User.count_transfers: return len(transfers) ``` You receive a Chalk DataFrame, which supports projections, filtering, and aggregations, among other operations. ### Projections By default, Chalk will materialize all scalar features on the Transfer feature class before calling your resolver. As an optimization hint, you can specify which features from the transfers that you'd like Chalk to materialize before calling the function. For example, if there were expensive features to compute on the transfer, you could scope the features to only the set you need: ``` @online def fn(transfers: User.transfers[Transfer.amount, Transfer.memo]) -> ...: transfers[Transfer.amount].sum() # Ok transfers[Transfer.from_institution] # Error: filtered out above ``` The error above is surfaced statically by our editor plugin. ### Filtering You can apply filters to the has-many inputs of resolvers: ``` @online def fn(transfers: User.transfers[Transfer.amount > 100]) -> ...: ``` Filters can be composed with projections following the semantics of the Chalk DataFrame. ``` @online def fn(transfers: User.transfers[Transfer.amount > 100, Transfer.memo]) -> ...: ``` Filters can also be used on features through has-one relationships. ``` @online def fn(transfers: User.transfers[Transfer.amount, Transfer.bank.location == "USA"]) -> ...: ``` More on projection and filtering here ### Has-many through has-one Has-many relationships can be required through has-one relationships: ``` @online def fn(transfers: User.account.transfers) -> ...: ``` As with scalar has-many dependencies, you can scope down the scalar features on the transfer to only those required: ``` @online def fn(transfers: User.account.transfers[Transfer.amount]) -> ...: transfers[Transfer.amount].sum() # Ok transfers[Transfer.from_institution] # Error: filtered out above ``` ### Has-many through optional-has-one If the has-one relationship that you're traversing is optional, then the transfers argument in the example above will either be None or a Chalk DataFrame. ### Has-one through has-many You can also select columns through nested has-one relationships that would not normally materialize. ``` @online def fn(transfers: User.transfers[Transfer.amount, Transfer.bank.name]) -> ...: ``` In the above example, if there was a has-one relationship between Transfer and Bank, we can fetch any scalar feature from Bank as well for the DataFrame. Note that the Bank.name column would not materialize with the simple transfers: User.transfers typing, since this typing only materializes scalar features in the root namespace. ### Has-many through has-many Has-many relationships can be required through other has-many relationships. For example, consider the following feature definitions for User, Account, and Transaction, where a user can have many accounts, each with many transactions. ``` from chalk.features import features, has_many, DataFrame @features class Transaction: id: str account_id: "Account.id" amount: float @features class Account: id: str user_id: "User.id" transactions: DataFrame[Transaction] @features class User: id: str total_spent: float accounts: DataFrame[Account] ``` We can resolve the total_spent feature on User by computing the sum of transaction amounts across all of a user's accounts, as shown below. ``` @online def get_total_spent( txns: User.accounts.transactions[Transaction.amount] ) -> User.total_spent: return txns.sum() ``` ### Time dependencies To set a time-based dependency in your resolver, you can use the Now keyword. Resolvers with a dependency on Now can use the input times passed in at query-time to compute either scalar or DataFrame outputs. ``` from chalk import online, DataFrame, Now from chalk.features import features from datetime import datetime @features class Account: id: str balance: float updated: datetime @online def get_daily_balance(now: Now) -> DataFrame[Account]: return DataFrame([ Account(id="1", balance=100, updated=now), Account(id="2", balance=200, updated=now), ]) ``` Read more about time-based dependencies here. ### Outputs Python resolvers can output either scalar features or a DataFrame of features. ### Scalar output ### Single output To return a single feature from a resolver, set the return type annotation to the feature you want to resolve: ``` from chalk.features import features @features class User: id: int name: str employer: str @online def resolve(u: User.id) -> User.name: return "Jennifer Doudna" ``` Equivalently, you can wrap the return value in the User class: ``` @online def resolve(u: User.id) -> User.name: - return "Jennifer Doudna" + return User(name="Jennifer Doudna") ``` ### Multiple outputs To return multiple features, return an instance of the feature class. In the type signature, specify the Features[...] class, parameterized by the features that you pass to the feature class. ``` @online def resolve(u: User.id) -> Features[User.name, User.employer]: return User( name="Jennifer Doudna", employer="University of California, Berkeley" ) ``` You only need to pass a subset of the features to the constructor for the feature class. The editor plugin will check that the type annotation you assign to the resolver matches subset of features passed to the constructor of the feature class. ### All features To return all features of a class, use Features[...] around the feature class. ``` @online def get_user(u: User.id) -> Features[User]: return User( name="Jennifer Doudna", employer="University of California, Berkeley" ) ``` If your resolver takes input features, those features are not considered as part of the output features. Note that the id feature is not returned from the function. This definition is equivalent to: ``` @online - def get_user(u: User.id) -> Features[User]: + def get_user(u: User.id) -> Features[User.name, User.employer]: return User( name="Jennifer Doudna", employer="University of California, Berkeley" ) ``` However, you may want to return almost all features of a class. Writing out all the features can be tedious and error-prone. You can subtract features from a feature class using the - operator: ``` from chalk.features import Features, ... @online def get_all_users(id: User.id) -> Features[User] - User.name: return User(employer="University of California, Berkeley") ``` Here, both the id feature and the name feature are not returned, which leaves only the employer feature. ### DataFrame output You can also output many instances of a feature class from a resolver by specifying a DataFrame as the return type of the function: ``` @offline def get_events() -> DataFrame[Transfer.uuid, Transfer.amount, Transfer.ts]: return DataFrame.read_csv(...) ``` Say you wanted to return many instances of a feature class, including nested features, from a resolver, then you can the DataFrame class for your return type and in your resolver definition. ``` @online def get_user_employer_information(id: User.id) -> DataFrame[ User.id, User.name, User.employer.name, User.employer.category ]: return DataFrame([ User( id="1", name="Jennifer Doudna", employer=Employer( name="University of California, Berkeley", category="Education" ) ) ]) ``` For more info on how to load batch data, see the Data Sources sections. DataFrame-returning resolvers don't need inputs. ### All features To return all features of a class in a DataFrame, use DataFrame[...] class around the feature class: ``` @online def get_all_users() -> DataFrame[User]: return DataFrame([ User( name="Jennifer Doudna", employer="University of California, Berkeley" ) ]) ``` ### Other DataFrame-returning resolvers Imagine a scalar feature you'd like to backfill over many thousands of primary keys and historical times. DataFrame-returning resolvers can dramatically reduce the computation time due to its vectorized handling. ``` @offline def get_new_feature_as_dataframe( df: DataFrame[Transaction.id, Transaction.amount] ) -> DataFrame[Transaction.id, Transaction.is_large]: return df.with_column(...) ``` The above resolver runs faster on a thousand rows than the equivalent scalar resolver ran a thousand times. Chalk also supports relationship-returning resolvers that enable users to return a DataFrame belonging to a has-many relationship. ``` @offline def relationship_returning_resolver( df: User.transactions[Transaction.id, Transaction.amount, Transaction.description], user_type: User.type ) -> User.transactions[Transaction.id, Transaction.transaction_type]: return ... ``` Just make sure that the return DataFrames do not have duplicate rows. That means no two rows should have the same primary key, or primary key & timestamp combinations if the feature time is also returned. ### Testing To test your Python resolvers, you can set up unit tests, construct integration tests or iterate on a branch. ### Examples ### ML Flow Python resolvers can also be used to load and run ML models. An example using ML Flow is shown below. This example loads a model from the ML Flow registry and uses it to make predictions. Using the @before_all decorator, the model is loaded once before any online requests are made. The model is then used in the get_prediction function to make predictions based on the input features. ``` import mlflow from chalk import online from chalk.features import before_all from src.models import FeatureClass import ExampleModelClass model: ExampleModelClass | None = None @before_all def load_model(): mlflow.set_registry_uri("EXAMPLE_REGISTRY_URI") mlflow.set_tracking_uri("EXAMPLE_TRACKING_URI") global model model = mlflow.load_model("models:/mymodel@production") @online def get_prediction( model_input: FeatureClass.model_input, ) -> FeatureClass.model_output: x_in = [model_input] return model.predict(x_in) ``` # Configuring Data Sources source: https://docs.chalk.ai/docs/configuring-data-sources ## Configure connections between Chalk and your data sources using the Chalk dashboard. Chalk is capable of integrating seamlessly with a variety of data sources. This allows you to query your data directly from Chalk, making it fast and easy to call external APIs, query production databases, and more. Data sources can be configured through our data source api or through the Chalk dashboard. In this section, we will cover how to configure data sources through the Chalk dashboard. ### Adding Data Sources Through the Dashboard Connections between Chalk and your data sources can be configured in the Chalk dashboard under Integrations > Datasources. The Add a data source button will allow you to select a data source type and configure the connection. A list of supported data sources can be found in the Integrations section. Once you have selected the data source you would like to integrate with, you will be prompted to fill in the necessary connection information. This may include credentials, host information, and other relevant details. ### Example: Adding a Kinesis Data Source In this example, we will show how to connect Amazon Kinesis to Chalk as a streaming data source. After selecting Kinesis as the data source type, you will be prompted to fill in the necessary connection information as well as information about the workload role that requires permissions to access your data source. Adding a Kinesis data source When setting up your data source, you also have the option to test the connection to ensure that everything is configured correctly by selecting the Test Data Source button. This will attempt to connect Chalk to your data source using the provided information and will alert you if there are any issues with the connection. In the case of Kinesis, you will be able to see messages from your stream if the test is successful: Testing a Kinesis data source connection ### Using your Data Source Once you have added your data source and successfully tested the connection, you can reference it in your resolvers. Integration examples can be found in the Integrations section or the section for your specific data source. # Unit Tests source: https://docs.chalk.ai/docs/unit-tests ## Unit tests for Chalk resolvers Chalk lets you specify your feature pipelines using idiomatic Python. You can unit test individual resolvers and combinations of resolvers as you would normal Python functions. Read more about how to write integration tests for resolvers here. ### Testing resolvers Consider the following features and resolvers: ``` from chalk.features import features, Features from chalk import online @features class Home: id: str address: str price: int sq_ft: int @online def get_address(hid: Home.id) -> Home.address: return "Bridge Street" if hid == 1 else "Filbert Street" @online def get_home_data( hid: Home.id, ) -> Features[Home.price, Home.sq_ft]: return Home( price=200_000, sq_ft=2_000, ) ``` You can test them just like normal Python functions using any unit testing framework: ``` def test_single_output(self): assert get_address(2) == "Filbert Street" def test_multiple_output(self): result = get_home_data(2) assert result.price == 200_000 assert result.sq_ft == 2_000 assert result == Home( price=200_000, sq_ft=2_000, ) ``` ### DataFrame inputs If you specify projections or filters in DataFrame arguments of resolvers, Chalk will automatically project out columns and filter rows in the input data. ### Filters Consider if we extend the Home class to include a rooms field: ``` + @features + class Room: + id: str + home_id: "Home.id" + name: str @features class Home: id: str address: str price: int sq_ft: int + rooms: DataFrame[Room] + num_bedrooms: int + + @online + def get_num_bedrooms( + rooms: Home.rooms[Room.name == 'bedroom'] + ) -> Home.num_bedrooms: + return len(rooms) ``` The get_num_bedrooms resolver filters the rooms argument to only include bedrooms. We can test this by passing in a list of rooms, some of which are bedrooms and some of which are not: ``` def test_get_num_rooms(): # Rooms is automatically converted to a `DataFrame` rooms = [ Room(id=1, name="bedroom"), Room(id=2, name="kitchen"), Room(id=3, name="bedroom"), ] # The kitchen room is filtered out assert get_num_bedrooms(rooms) == 2 # `get_num_bedrooms` also works with a `DataFrame` assert get_num_bedrooms(DataFrame(rooms)) == 2 ``` Note that although we passed in a list of three rooms, only two of them were bedrooms, so the resolver returns 2. Furthermore, we didn't need to convert the list of rooms to a DataFrame. In this case, we passed in a list of Room objects, but Chalk automatically converts it to a DataFrame for us. We also could have passed in a polars.DataFrame or any other valid constructor for DataFrame. ### Projections Chalk will also automatically project the input data as specified by the DataFrame argument. So, if you specify which columns you need in the resolver body, but accidentally use an extra column in your resolver body, your unit tests will fail. For example, if we wrote a resolver for summing the square footage of all rooms in a home using the Room.sq_ft feature, but accidentally excluded the Room.sq_ft column in the DataFrame argument: ``` @online def get_sqft( rooms: Home.rooms[Room.name] ) -> Home.sq_ft: # This will fail because we used `Room.sq_ft` return rooms[Room.sq_ft].sum() ``` Then our unit tests will fail: ``` def test_get_sqft(): rooms = [ Room(id=1, name="bedroom", sq_ft=100), Room(id=2, name="kitchen", sq_ft=200), Room(id=3, name="bedroom", sq_ft=300), ] # This will fail because we used `Room.sq_ft`, # which is excluded by the `Home.rooms[Room.name]` # argument assert get_sqft(rooms) == 400 ``` ### after(...)/before(...) and time filtering Some DataFrame filters are implicitly resolved relative to "now". In offline_query and online_query, this value is controlled by input_times= and now= parameters, respectively. In unit tests, the value defaults to datetime.now(), but can be explicitly set with the chalk.freeze_time context manager: ``` from chalk import freeze_time @online def get_p30d_transactions_count(txns: User.transactions[after(days_ago=30)]) -> User.recent_txn_count_30d: """ Count the number of transactions that have occurred in the past 30 days """ return txns.count() NOW = datetime.now(tz=timezone.utc) transactions = DataFrame([ Transaction(id=1, created_at=NOW), Transaction(id=2, created_at=NOW - timedelta(days_ago=30)), Transaction(id=3, created_at=NOW - timedelta(days_ago=32)) ]) # There is a single transaction in the range (30, 0] days ago assert get_p30d_transactions_count(transactions) == 1 # There are two transactions in the range (45, 15] days ago with freeze_time(at=NOW - timedelta(days_ago=15)): assert get_p30d_transactions_count(transactions) == 2 ``` Note that the at= parameter must be timezone-aware. # Time Overview source: https://docs.chalk.ai/docs/time ## Manage feature timestamps and point-in-time query timestamps. ### Overview Chalk uses two main timestamps you should be aware of as you build your Chalk project: - Feature time (FeatureTime): The time at which a feature was observed, which is used in query time filters and aggregation time windows. - Query time (Now): The time a query should assume is "now" when retrieving data. Features whose feature time comes after a given query time will never be returned for those queries, in online or offline contexts. Feature time is returned in the __observed_at__ column in Chalk query results. Query time is returned in the __ts__ column. ### Feature time Feature time is the time at which a feature was observed. By default, Chalk sets a feature's time to the feature's resolver execution time. The feature time can be overridden for a feature class, accessed from resolver parameters, and requested in query inputs and outputs. You may have multiple timestamps associated with your data. It's important to set the feature time to the value that most closely represents when your system would have accessed the data in production. For example, in an asynchronous streaming system, you may have one timestamp for when an event was added to your task queue and another timestamp for when the event was removed from the queue and processed. We recommend using the latter timestamp as your feature time to make your training most closely resemble production. If you use the timestamp for when an event was added to your task queue, you would be training your system with data it would not have been able to access in production. ### Setting feature time for a feature class Each individual feature has its own feature time, which is used to retrieve point-in-time correct data for temporal consistency. To access the latest time associated with any feature in a feature class, use the special FeatureTime feature throughout Chalk. By default, if it is present, Chalk will treat a feature named ts with datetime.datetime type as the FeatureTime value. Otherwise, you can use the FeatureTime type annotation to set a different name: ``` from chalk.features import features, FeatureTime @features class User: id: int name: str timestamp: FeatureTime # datetime.datetime under the hood ``` Using your FeatureTime feature, you can access and override feature time for the whole feature class. You can directly set the FeatureTime value by returning it from a resolver: ``` @offline def fn(...) -> Features[User.name, User.timestamp]: return User( name="Maryam Mirzakhani", ts=datetime(2014, 8, 12, tzinfo=timezone.utc) ) ``` You can include the FeatureTime feature in query output. Its value will be set to the maximum timestamp across all features in its feature class. ### Time filtering in resolvers has-many features create DataFrames. These DataFrames can be filtered with before and after. Regardless of which time filters you use, Chalk will never return features where the feature time is strictly greater than the current query time (Now), in order to maintain temporal consistency. ### Examples To compute the number of transfers a user made in the last seven days, use after: ``` from chalk.features import after, ... @online def fn(transfers: User.transfers[after(days_ago=7)]) -> ...: return transfers.count() ``` To compute the number of transfers a user made more than seven days ago, use before: ``` from chalk.features import before, ... @online def fn(transfers: User.transfers[before(days_ago=7)]) -> ...: return transfers.count() ``` Combine before and after to retrieve transfers made 1-2 weeks ago: ``` from chalk.features import before, after, ... @online def fn(transfers: User.transfers[after(days_ago=14), before(days_ago=7)]) -> ...: return transfers.count() ``` All of these examples can be used in combination with other DataFrame projections and filters. You may also find windowed aggregations useful. ### Interaction with the online store Features with overridden observation timestamps are treated specially when inserted into the online store. In particular, Chalk will always check for existing "newer" feature values in the online store before inserting historically dated feature values. This means that you can safely ingest large quantities of backdated features without accidentally ingesting stale data into the online store. Additionally, once features are inserted into the online store, Chalk tracks the source observation timestamps when these feature values are returned as part of online queries. Chalk uses these source timestamps to compute the "feature staleness" metric. Staleness in this context is defined as "query time - observation time". ### Interaction with the offline store Features with overridden observation timestamps are inserted into the offline store with the timestamp that you specify. The observation timestamp works like an "effective as of" timestamp, so if you insert something like this: ``` | id | feature | value | timestamp | |---------------------------------------------| | 1 | age | 7 | 2022-02-01T00:00:00Z | ``` into an offline store that already contained these observations: ``` | id | feature | value | timestamp | |---------------------------------------------| | 1 | age | 6 | 2022-01-01T00:00:00Z | | 1 | age | 8 | 2022-03-01T00:00:00Z | ``` then the observation will be interleaved "in between" the existing observations, and you would see the following query results: ``` id, age, <= 2022-02-01 output: 7 id, age, <= 2022-03-02 output: 8 id, age, <= 2022-01-02 output: 6 ``` ### Enforcing a TTL Features in the offline store have an optional TTL (time to live). When a feature has a TTL value, it will never be returned at any time later than FeatureTime + TTL. For example, you may not want to consider credit scores which were retrieved more than a year ago. Setting offline_ttl will make credit_score return None if last observed credit score is more than one year old in comparison to the current query time. ``` @features class User: id: str credit_score: int = feature(offline_ttl=timedelta(years=1)) ``` ### Query time Query time is the time treated as "now" within a query context. For online queries, Now is equal to datetime.now(). For offline queries, you can pass one or more timestamps that will be used as the query time for each input row. In Chalk Expressions, you can reference the query time with _.chalk_now. ### Setting query time In training, you will likely want to retrieve data as if you are at a point in the past to create the most accurate predictions. We cover this idea in greater detail in our temporal consistency documentation. To set the query's "now" time, pass input_times as either a single timestamp or as a list corresponding to the Now times to use for each entry in input: ``` from datetime import timezone ChalkClient().offline_query( # Pass id 1 multiple times because we want to # request it with multiple input_times input={User.id: [1, 1, 1]}, input_times=[ datetime.now(tz=timezone.utc) - timedelta(days=365 * 10), datetime.now(tz=timezone.utc) - timedelta(days=365), datetime.now(tz=timezone.utc) - timedelta(days=0), ], output=[User.age_in_years], ) ## Output: # | id | age_in_years | # | 1 | - 10 | # | 1 | - 1 | # | 1 | - 0 | ``` ### Accessing query time in resolvers To access the query time in your resolvers, you can reference a special feature called Now, which is a datetime.datetime object. You can pass Now in Python resolvers: ``` from chalk import Now @online def get_age_in_years(birthday: User.birthday, now: Now) -> User.age_in_years: return (now - birthday).years ``` Now can be used in DataFrame resolvers as well in order to compute bulk values: ``` @online def batch_get_age_in_years(df: DataFrame[User.id, User.birthday, Now]) -> DataFrame[User.id, User.age_in_years]: return ( df.to_polars() .select( pl.col(User.id), pl.col(str(User.birthday) - pl.col(str(Now))).alias(str(User.age_in_years)) ) ) ``` You can also reference ${now} in SQL file resolvers. ``` -- source: sql_file_resolver_temp_db -- resolves: tv_episode select id, name, season_no, episode_no, show_name, air_date from tv_episodes where air_date < ${now} and id = ${tv_episode.id} ``` You can also reference _.chalk_now in Chalk Expressions, particularly in windowed aggregations: ``` from chalk import features, _ from chalk.streams import windowed, Windowed @features class Store: id: int visits: DataFrame[Visit] num_recent_visits: Windowed[int] = windowed( "1d", "7d", "30d", expression=_.visits[ _.ts < _.chalk_now, _.ts > _.chalk_window ].count(), default=0, ) ``` ### Accessing query time in query results Chalk Datasets return the query time in the __ts__ column. When converting Chalk Datasets to Polars or Pandas DataFrames, you may want to include the query time column. To do so, pass output_ts to your to_polars or to_pandas calls. You may pass a column name to output_ts to set the name of the query time column. If you pass True, the query time column name will be __chalk__.CHALK_TS. Be careful to not mix up ts and ts. ts represents the query time, or the time the query treats as "now" during query execution. ts is a common name for the feature representing FeatureTime, the time at which a feature was observed. ### Timezone handling for naive datetimes Chalk stores UTC as the timezone for naive datetime objects. Additionally, Chalk assumes UTC if retrieving naive datetimes from data stores. We recommend that you include timezone information on all datetime objects you work with to avoid ambiguity. ### When to use FeatureTime vs Now Chalk enables different levels of temporal consistency, which also have different impacts on performance. When a resolver does not have a FeatureTime or Now as an input or output, there is no temporal pushdown filtering, so the resolver will be fully time independent, which will also be more performant. If the resolver has a FeatureTime as an input, then Chalk will apply a temporal pushdown filter. Then, if a resolver takes Now as an input, the resolvers can fully interact with time, which will provide the most accuracy for point-in-time feature resolution but also be the least performant. # Temporal Consistency source: https://docs.chalk.ai/docs/temporal-consistency ## Point-in-time queries for training data sets. ### Introduction Temporal consistency is crucial for ensuring your model's training performance is reflective of production. When creating datasets for model training, your system must consistently retrieve data as it would appear at a specific point in time. If your model trains on data that was retrieved after the point where it would have made a prediction, it will not perform consistently in production. Chalk performs point-in-time lookups on your training data so you can train your models knowing they won't receive data from the future, even across complex relationships. ### Scenario 1: Historical loan decisions In this scenario, we will consider a machine learning model for determining whether to issue loans to businesses based on their financial performance. For each business, we track sales and COGS (cost of goods sold) over time in millions of dollars. We define a Business feature class with sales and cogs features: ``` from chalk.features import features, FeatureTime @features class Business: id: int sales: float cogs: float ts: FeatureTime ``` After collecting historical data on the performance of businesses we previously considered, we want to retrain our model. For a business with id=123, we gave loans at t1 and t2: In training, you want to know the observed gross profit and COGS for the business at the time you made the loan without knowing the future values of those features. For example, at t1, when we issued the loan, we had observed sales of $1.3M. It would be inconsistent for this model to train with the $1M data point because this data would not have existed at t1. At t1 and t2, the temporally consistent values of sales and cogs would be as follows: | Feature | Value at `t1` | Value at `t2` | | -------------- | ------------- | ------------- | | Business.sales | 1.3 | 1 | | Business.cogs | 0.5 | 0.4 | Each of these values occurred at or before the sample time and is valid to use in training. ### Sample code Chalk allows you to control the time your query considers to be "now", which is covered in greater detail in our time documentation. To set the query's "now" time, pass input_times to your offline queries: ``` from chalk.client import ChalkClient t1 = datetime.now() - timedelta(days=365) t2 = datetime.now() - timedelta(days=30) dataset = ChalkClient().offline_query( input={ # Sample business 123 twice because we have two input_times Business.id: [123, 123], }, # Each element of `input_times` corresponds to the element # with the same index in `input` input_times=[t1, t2], # Sample all features of business. # Alternatively, sample only the features you need: # output=[Business.sales, Business.cogs] output=[Business], ) ``` This query will return a Dataset with temporally consistent values for each input time. ### Scenario 2: Backfilling new features Temporal consistency is especially difficult when you want to build new features. Building on the first scenario, imagine you've observed Business.sales and Business.cogs many times in the past, and for each of the businesses that you track, these values have changed over time. Now, you want to compute a new feature, Business.gross_profit, which is the difference between Business.sales and Business.cogs. We can add an expression to compute this new feature: ``` - from chalk.features import features, FeatureTime + from chalk.features import _, features, FeatureTime @features class Business: id: int sales: float cogs: float ts: FeatureTime + gross_profit: float = _.sales - _.cogs ``` With the same historical data from the previous scenario, we can determine the correct inputs for computing Business.gross_profit: Business.gross_profit at t1 and t2 would be computed with the following values: | Query time | `Business.sales` | `Business.cogs` | `Business.gross_profit` | | ---------- | ---------------- | --------------- | ----------------------- | | `t1` | 1.3 | 0.5 | 0.8 | | `t2` | 1.0 | 0.4 | 0.6 | In your code, you can compute historical values for updated features by running an offline query with recompute_features=True. You may also consider one of our other backfill options for other use cases. As you start layering more resolvers or using has-many relationships, temporal consistency becomes even more difficult to reason about. Chalk manages the complexity under the hood so that you can write any query and expect temporally consistent results. # Feature Development Lifecycle source: https://docs.chalk.ai/docs/lifecycle ## Developing, testing, deploying, and backfilling a new feature Chalk is the fastest way to deploy new features and feature pipelines to production. This demo covers creating a new feature. You'll develop, unit test, integration test, and deploy a resolver for this feature. Then you'll see how to backfill that feature to Chalk's offline store so that data scientists can use it in historical training sets. ### Develop The first step is to write a new feature and resolver for that feature. Imagine you wanted to create a new feature called user.name_email_match_score. This feature should capture the similarity between a user's name and email. Put another way: andy and andy@gmail.com should produce a high score, and emily and andy@gmail.com should produce a low score. The first step is to add the new feature, and a resolver for that feature. Our new resolver name_email_match_scorer takes a dependency on the user's email and fullname in the argument list. Then, it declares that it returns the User.name_email_match_score in the return type signature. In the body of the function, we compute the Jaccard index between the email without the domain and the full name. ### Test Now that you've written the new feature and resolver, it's time to validate your change. Chalk supports both unit testing and integration testing of your feature pipelines. ### Unit test You can unit test resolvers just like normal Python functions using any unit testing framework: ``` from pytest import approx def test_name_email_match_scorer(): assert approx(0.60) == name_email_match_scorer( "katherine.johnson@nasa.gov", "Katherine Johnson", ) assert approx(0.39) == name_email_match_scorer( "katherine.johnson@nasa.gov", "Eleanor Roosevelt", ) ``` Here, the Jaccard index for katherine.johnson@nasa.gov is higher with the name Katherine Johnson than with the name Eleanor Roosevelt, as expected. ### Deploy branch Now that the unit tests have passed, you can create a Branch Deploy with the new changes. Chalk allows you to create an unlimited number of branch deployments. Branch deployments run all of your resolvers in the same way that they run in production. However, branch deployments don't impact the offline store. You can create a branch deployment with the following command: ``` chalk apply --branch test ``` ### Integration test With the branch deployment created, you can run integration tests on your changes. ``` import pytest from chalk.client import ChalkClient @pytest.fixture(scope="module") def chalk_client(): client = ChalkClient(branch="test") return client def test_get_features(chalk_client): resp = chalk_client.query( input={User.id: 1}, output=[ User.credit_report.fico_score, User.email, User.name, User.dob, ], ) assert resp.get_feature_value(User.credit_report.fico_score) == 700 assert resp.get_feature_value(User.email) == "katherine.johnson@nasa.gov" ``` Here, the Chalk API client is configured to use the preview deployment id returned in the previous step. ### Deploy Once you've tested your changes, it's time to deploy! This step looks much like the preview deployment, but this time without the --branch flag: ``` chalk apply ``` Now, your production environments can request the new user.name_email_match_score feature. ``` chalk query --in user.id=1 --out user.name_email_match_score ``` ### Backfill The user.name_email_match_score feature is live! But historically, this feature did not exist, and you won't be able to sample its values at previous times until you backfill the values, which you can do with the 'chalk trigger' command. ``` chalk trigger --resolver example.name_email_match_scorer --lower-bound 2020-05-05T12:00:00+00:00 --persist-offline=True ``` This command will backfill historical values for the user.name_email_match_score feature. ### Offline training With the feature backfilled, you can query the historical value: That's how you deploy a new feature with Chalk! Let a member of our technical team know if we can be helpful. ### Delete Mistakes are inevitable, but Chalk provides the tools you need to quickly and easily recover and keep going. ### Drop a feature If you need to change a feature's type (for example from string to int), or if you want to drop all the data for a feature value, the Chalk CLI has you covered with chalk migrate-type. Simply execute the command from the CLI and you'll have a fresh start to recreate the feature and its data as necessary. ### Delete a row Sometimes you just need to fully remove a record from your systems, whether because of a GDPR mandated "right to forget" request or due to a business requirement. Chalk provides the chalk delete command to meet this need. # Working With Branches source: https://docs.chalk.ai/docs/branches ## Make rapid changes and explore features with branches. Branch deployments allow users to quickly iterate on features and datasets.When working in a branch users have a number of useful capabilities, including: - Chalk can automatically watch your files and deploy any changes - Updates to your project are deployed in seconds - Adjust and create features and resolvers inside a Jupyter notebook - Quickly recompute datasets to see the effects of changes - Branches have a consistent name across deployments However, branch deployments do not pick up changes to environment variables or secrets, so these changes would require a new non-branch deployment to take effect. ### Create a Branch Deployment To create a branch, simply run chalk apply --branch . Chalk will create a new branch in the environment that you can interact with. Now, you can make queries against your branch with chalk query --branch.If you're using the Python API, you can set the branch when creating your client, and all subsequent commands will execute against your branch. ### Continuously Deploy Changes You can ask the Chalk CLI to continuously deploy changes to your branch using the watch flag. Chalk will automatically keep the branch up to date as you make changes locally. ### Develop in a Notebook One of the major advantages of branch deployments is the flexibility they offer when working in notebooks.Once you've deployed a branch, you can iteratively edit features and resolvers and see the effects of these updates in a dataset. Create a chalk client and set the branch parameter equal to your branch. Then, any time you execute a cell that contains a class annotated with @features it will automatically be updated in your branch. Similarly, executing cells that contain a function annotated with @online or @offline will automatically deploy the resolver to your branch. ``` @features class NewFeatures: id: int name: str greeting: str @online def new_resolver(name: NewFeatures.name) -> NewFeatures.greeting: return f"Hello {name}!" ``` ### Recompute Datasets As you adjust features and resolvers, you can iteratively see how your changes affect your feature values with the Dataset.recompute method. Just pass any features you want to be re-computed as arguments to the features parameter, and Chalk will generate a new Dataset revision using the latest definitions of features and resolvers. When used in conjunction with the ability to adjust your features and resolvers in the notebook, this tool allows developers and data scientists to rapidly experiment and productionize their work. ### Example In the following example we add an oversize feature to an existing dataset of loans. We start out with a dataset. We may have produced this dataset manually, by calling an external API, or by executing offline_query. ``` dataset shape: (786, 3) ┌───────────────┬──────────────┬─────────────────────────┐ │ loan.amount ┆ loan.id ┆ loan.event_time │ │ --- ┆ --- ┆ --- │ │ f64 ┆ str ┆ datetime[μs, UTC] │ ╞═══════════════╪══════════════╪═════════════════════════╡ │ 165435.647396 ┆ l_G3Mc6bi9y4 ┆ 2023-04-08 10:03:09 UTC │ │ 405006.796909 ┆ l_O9OK3us7t2 ┆ 2022-02-07 20:46:08 UTC │ │ 680377.427817 ┆ l_L0Gg0Bd1v3 ┆ 2023-02-16 16:48:29 UTC │ │ 562678.545344 ┆ l_D1Rb5Jq4U5 ┆ 2022-06-21 06:15:19 UTC │ │ … ┆ … ┆ … │ │ 750583.279013 ┆ l_W4ZY9OK5N7 ┆ 2021-12-16 21:40:27 UTC │ │ 71698.15609 ┆ l_H9tG2yJ5B6 ┆ 2023-01-04 11:33:54 UTC │ │ 488697.890372 ┆ l_L4fd4xu4w2 ┆ 2022-08-22 03:55:16 UTC │ │ 769665.198436 ┆ l_k4Dq8bl7b3 ┆ 2022-10-31 07:15:52 UTC │ └───────────────┴──────────────┴─────────────────────────┘ ``` In our Jupyter notebook, we can execute this in a cell to set up our notebook to point to our branch, then update our branch with the new feature and resolver. ``` client = ChalkClient(branch="") # We want to introduce an `oversize` detection feature @features class Loan: id: str event_time: datetime status: str amount: float oversize: bool # Detect any loans bigger than $250,000 @online def is_oversize(amt: Loan.amount) -> Loan.oversize: return amt > 250000 ``` Finally, we can recompute our dataset, telling it to calculate Loan.oversize, to get a dataset back with our new feature values. ``` # Recomputing the dataset will add the oversize column to our result dataset.recompute(features=[Loan.oversize]) shape: (786, 4) ┌───────────────┬───────────────┬──────────────┬─────────────────────────┐ │ loan.oversize ┆ loan.amount ┆ loan.id ┆ loan.event_time │ │ --- ┆ --- ┆ --- ┆ --- │ │ bool ┆ f64 ┆ str ┆ datetime[μs, UTC] │ ╞═══════════════╪═══════════════╪══════════════╪═════════════════════════╡ │ true ┆ 165435.647396 ┆ l_G3Mc6bi9y4 ┆ 2023-04-08 10:03:09 UTC │ │ true ┆ 405006.796909 ┆ l_O9OK3us7t2 ┆ 2022-02-07 20:46:08 UTC │ │ true ┆ 680377.427817 ┆ l_L0Gg0Bd1v3 ┆ 2023-02-16 16:48:29 UTC │ │ true ┆ 562678.545344 ┆ l_D1Rb5Jq4U5 ┆ 2022-06-21 06:15:19 UTC │ │ … ┆ … ┆ … ┆ … │ │ true ┆ 750583.279013 ┆ l_W4ZY9OK5N7 ┆ 2021-12-16 21:40:27 UTC │ │ true ┆ 71698.15609 ┆ l_H9tG2yJ5B6 ┆ 2023-01-04 11:33:54 UTC │ │ true ┆ 488697.890372 ┆ l_L4fd4xu4w2 ┆ 2022-08-22 03:55:16 UTC │ │ true ┆ 769665.198436 ┆ l_k4Dq8bl7b3 ┆ 2022-10-31 07:15:52 UTC │ └───────────────┴───────────────┴──────────────┴─────────────────────────┘ ``` # Queries Overview source: https://docs.chalk.ai/docs/query-overview ## Fetch feature values via queries. To request or compute data with Chalk, you'll use queries. In general, when you run a Chalk query, you are either requesting the most up-to-date values for your feature classes, requesting a set of historical data for your feature classes, or running a backfill or batch job. The first use case is accomplished through online queries, which try to return values for a single feature class as quickly as possible, taking advantage of caching and distributed execution. The latter use cases are accomplished through offline queries which use the offline store and can return multiple instances of a feature class for multiple primary keys or timepoints. ### Running queries At a high level, a query specifies input features and output features. Inputs differ slightly for online queries and offline queries, but in both cases the input must contain the primary key of your requested feature class. ### Running online queries Online queries can be run using query: ``` chalk query --in user.id=1 --out user.name ``` or one of our API Clients: ``` from chalk.client import ChalkClient client = ChalkClient() client.query( input={'user.id': 1}, output=['user.name'], # branch='test', # run against a branch ) ``` ### Running offline queries Offline queries can be run with one of our API Clients: ``` from chalk.client import ChalkClient from datetime import datetime client = ChalkClient() client.offline_query( input={'user.id': [1,2,3,4]}, output=['user.name'] # branch='test', # run against a branch # recompute_features=True, # recompute features # run_asynchronously=True, # run in separate pod from active deployment # max_samples = 10, # max of 10 samples # lower_bound=datetime(2024, 10, 12), # sample computed after 10.12.2024 # upper_bound=datetime(2024, 10, 20), # sample computed before 10.20.2024 ) ``` ### Running queries using gRPC If you have a gRPC query server active in your environment, you can also run queries using the gRPC client: ``` from chalk.client.client_grpc import ChalkGRPCClient grpc_client = ChalkGRPCClient() grpc_client.query( input={'user.id': 1}, output=['user.name'] ) ``` ### Scheduled and triggered resolver runs Specific resolvers can also be scheduled or triggered (for instance, as part of pipelines like Airflow). Specific queries can also be scheduled with ScheduledQuery. Triggers and schedules are useful for pulling data from "slow" data sources into your offline and online store. ### Query side effects Chalk queries can also write data. This is an essential part of Chalk: every time you compute a feature through an online query, the output is written down in the offline store. This makes it easy to: - create datasets from your previously computed features, - monitor and track your computed features over time. Though not the default, offline queries can write data to both the offline store and the online store using etl_offline_to_online. This can be useful when backfilling data from slow data sources or when performing expensive feature computation that would otherwise significantly impact the latency of your online queries. ### Online and offline query differences | Online query | Offline query | | -------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Runs only @online resolvers | Runs both @online and @offline resolvers | | Returns one row of data about one entity | Returns a DataFrame of many rows of historical data corresponding to multiple entities point-in-time | | Designed to return data in milliseconds | Blocks until computation is complete, not designed for millisecond-level computation | | Queries the online store and calls `@online` resolvers for quick retrieval | Queries the offline store which stores all data from online queries, unless `recompute_features=True`, in which case `@offline` and `@online` resolvers are used to resolve the outputs | | Writes output data to online store database and offline store database | Writes output to a parquet file containing results to cloud storage. Only writes to online store or offline store if specified. | # Online Queries source: https://docs.chalk.ai/docs/query-online ## Fetch feature values via online queries Online queries access or compute feature values for a single feature set in real time. The term "real time" means that responses should seem practically instantaneous, even if they need to compute features on new data. However, online queries don't only perform data retrieval, they also store the results of the features that they compute. This provides visibility and long-term tracking into the features you are generating. In this section, we provide a high level overview of how online queries compute and record their outputs. ### How Do Online Queries Work? Chalk responds to online queries by getting and executing a query plan. ### Getting a Query Plan A query plan is a sequence of tasks, some of which can be executed in parallel, that will produce a target output. Although a simplification, your resolvers are a subset of these tasks. Consider the following feature class and resolvers: ``` from chalk.features import features, online @features class User: id: int name: str is_palindrome: str is_short: bool palindrome_and_short: bool @online def get_is_palindrome(name: User.name) -> User.is_palindrome: return name == name[::-1] @online def get_is_short(name: User.name) -> User.is_short: return len(name) < 3 @online def get_is_short_palindrome(is_short: User.is_short, is_palindrome: User.is_palindrome) -> User.is_short_palindrome: return is_short and is_palindrome ``` If you were to begin assembling a dependency graph for the features. You would wind up with something like the following: ``` ┌───────────┐ │name │ └┬─────────┬┘ ┌▽───────┐┌▽────────────┐ │is_short││is_palindrome│ └┬───────┘└┬────────────┘ ┌▽─────────▽────────┐ │is_short_palindrome│ └───────────────────┘ ``` When you run an online query, such as: ``` chalk query --in user.id=1 --in user.name=bob --out is_short_palindrome ``` Chalk constructs a plan for how to "solve" this query. This query plan is viable for other queries with the same input and output features. Running chalk query with the --explain flag outputs your query plan. The following query can reuse the plan generated by the one above: ``` chalk query --in user.id --in user.name=bartholomew --out is_short_palindrome ``` Even though both the input name and the output of the query are different, the query plan remains valid. ### Executing a Query Plan As illustrated above, a query plan is not a linear sequence of tasks that must be executed one after another: a lot of work can often be performed in parallel. After getting a query plan, Chalk distributes subtasks to workers, applies a number of optimizations on your resolvers/datasource connections, and computes the target outputs of your query. These outputs are then returned. There are multiple ways to run an online query. You can use the Chalk API Client, the Chalk CLI, or the REST API. For example the following calls all trigger the same online query: With the Chalk Python API Client: ``` from chalk.client import ChalkClient from src.models import User client = ChalkClient() results = client.query( inputs={ User.id: 1, }, outputs=[User.organization, User.name], ) ``` With the Chalk CLI: ``` chalk query --in user.id=1 --out user.organization --out user.name ``` With the REST API: ``` curl -X POST {YOUR_ENGINE_URL}/v1/query/online \ -H "Authorization: Bearer $access_token" \ -H 'x-chalk-env-id: {YOUR_ENV_ID}' \ -H 'x-chalk-deployment-type: engine' \ -H 'content-type:application/json' \ -d '{"inputs": {"user.id": 1}, "outputs":["user.organization", "user.name""]}' ``` ### Caching and Writing Online queries write computed values to two places: the offline store and the online store. However, computed features are only written to the online store if they have a caching policy. The online store is used to circumvent recomputation of expensive features that are either unlikely to have changed or can tolerate slightly stale values. Properly configuring the caching policies for your features can make your online queries significantly more efficient. If you haven't specified a caching policy, Chalk recomputes the values for a feature each time it is requested. We go into more depth on query caching here. # Chalk Clients source: https://docs.chalk.ai/docs/query-basics ## Fetch feature values via online query. Chalk maintains several client libraries (gRPC) and a REST API for fetching feature values. ### Library support Chalk maintains libraries in several major languages for fetching online feature values. If you need support for a language that we don't support, let us know! We also support a rest API if you'd like to build your own. ### REST API Chalk supports a REST API for querying online features and exposes this endpoint in several API clients. When you execute an online query, resolvers will execute to produce the requested data. Online query will prioritize running online resolvers over offline resolvers to compute features if both are possible. The following endpoint can also be hit with the python ChalkClient by using its query method. For information on how to authenticate the ChalkClient, check out the section on authentication. Read more about the parameters to this method here. ### Request Input features and values are provided at the time of request. For example, primary key-value pairs often designate the subset of data returned. Feature inputs are provided by fully qualified path. Has many features are input as lists, and struct features are input as JSON. An example of passing a user with two credit cards as input: Outputs are the features that you'd like to compute from the inputs. Maximum staleness overrides for any output features or intermediate features. See query caching for more information. The context object controls the environment and tags under which a request should execute resolvers: The environment in which to run the resolvers. Like resolvers, API tokens can be scoped to an environment. If no environment is specified in the query, but the token supports only a single environment, then that environment will be taken as the scope for executing the request. The tags used to scope the resolvers. The preview deployment id. See Preview Deployments for more information. The query name. See NamedQuery for more details. If specified, routes to the relevant branch. See Branches for more information. More information on parameters is available here ### Response The outputs features and any query metadata The name of the feature requested, eg. user.identity.has_voip_phone . The value of the requested feature. If an error was encountered in resolving this feature, this field will be empty. The error code encountered in resolving this feature. If no error occurred, this field is empty. Metadata pertaining to the feature, including the resolver run and whether the result was a cache hit. Errors encountered while running the resolvers. Each element in the list is a ChalkError. If no errors were encountered, this field is empty. Metadata related to the query. Returned if include_meta or explain is set to True . The time, expressed in seconds, that Chalk spent executing this query. The id of the deployment that served this query. The id of the environment that served this query. The short name of the environment that served this query. For example: "dev" or "prod". A unique ID generated and persisted by Chalk for this query. All computed features, metrics, and logs are associated with this ID. Your system can store this ID for audit and debugging workflows. At the start of query execution, Chalk computes 'datetime.now()'. This value is used to timestamp computed features. Deterministic hash of the 'structure' of the query. Queries that have the same input/output features will typically have the same hash; changes may be observed over time as we adjust implementation details. An unstructured string containing diagnostic information about the query execution. Only included if explain is set to True . ### Query Explanation Chalk offers support for the user for when queries don't work. The first step is always to check to see the response contains any errors. Often, the error message will directly point to the failure. In the case of more complicated queries, queries can be sent with explain=True. This will return a representation of the query plan in the meta return attribute. The user can use this information to verify the resolvers and operators ran during execution. Beware, this will result in slower execution times. Some queries that involve multiple operations might need additional tracking. Users can supply store_plan_stages=True to store intermediate outputs at all operations of the query. This will dramatically slow things down, so use wisely! These results are visible in the dashboard under the "Queries" page. For more information, read the ChalkClient docs here. ### Online Query Bulk Compute feature values for many rows of inputs using online resolvers. This endpoint is similar to the online query endpoint, but takes in lists of inputs and produces one output per row of inputs. This is appropriate when you want to fetch the same set of features for many different input primary keys. The following endpoint can be accessed with the Python ChalkClient by using its query_bulk method. ### Request The request body should be a binary payload containing: - A magic string identifier - Serialized query metadata - Feature data in Apache Feather format When using the ChalkClient, this serialization is handled automatically. For direct HTTP usage, the structure is: - Request inputs are provided as mappings of feature names to lists of values - Each list should have the same length, representing multiple rows of data Input features and lists of values. Each key is a feature name (e.g., "user.id") and each value is a list of values for that feature. All lists must have the same length, where each element represents one row. Has-many features are input as lists within each row element, and struct features (has-one) are input as JSON objects within each row element. Example with simple, has-one, and has-many features: ``` { "user.id": [1, 2], "user.name": ["Alice", "Bob"], "user.profile": [ {"age": 30, "city": "NYC"}, {"age": 25, "city": "SF"} ], "user.cards": [ [{"id": "card1"}, {"id": "card2"}], [{"id": "card3"}] ] } ``` The features that you'd like to compute from the inputs. Same as online query. List of timestamps (ISO format) for each row. If provided, the list must match the length of the input value lists. Each timestamp represents the query time for the corresponding row. Maximum staleness overrides for any output features or intermediate features. See query caching for more information. The context object controls the environment and tags under which requests execute: The environment in which to run the resolvers. The tags used to scope the resolvers. If specified, all required_resolver_tags must be present on a resolver for it to be eligible to execute. The semantic name for the query, e.g., "loan_application_model". See NamedQuery. The version of the named query to execute. A globally unique ID for the query, used in logs and web interfaces. If specified, routes to the relevant branch. If specified, routes to the relevant preview deployment. If true, returns query execution plan in the response metadata. Makes the query slower. If true, stores intermediate outputs at all query plan stages. Dramatically impacts performance. Arbitrary key-value pairs to associate with the query. An immutable context accessible from Python resolvers. See ChalkContext. ### Response The response is a binary payload in Apache Feather format containing the results. A list of query results, where each result contains: A DataFrame containing the scalar feature values for all rows. Each row corresponds to an input row. Columns are feature names, and values are the computed feature values. A map of feature names to DataFrames for has-many features. Errors encountered while running the resolvers. Metadata about query execution including execution duration, deployment ID, query ID, etc. See the online query response documentation for QueryMeta details. ### Offline Query Submit an offline query to compute feature values from the offline store or by running offline/online resolvers. Offline queries are typically used for generating training datasets and run asynchronously. The following endpoint can be accessed with the Python ChalkClient by using its offline_query method. See the offline query documentation for more information. ### Request The request body is a binary payload containing: - Serialized query plan (protobuf) - Query metadata (JSON) - Input data (Apache Feather format) When using the ChalkClient, this serialization is handled automatically. The features for which there are known values. Can be: - A mapping of feature names to lists of values (similar to bulk query format) - A DataFrame with input data - A URI pointing to input data in cloud storage When using a mapping, has-many features are input as lists within each row, and struct features (has-one) are input as JSON objects within each row. See the bulk query input format above for examples. Timestamps for point-in-time correctness. If a list, must match the length of input rows. See temporal consistency. The features to compute or sample. If a feature was never computed for a sample, its value will be null. Features that must exist in each row. Rows where a required output was never stored will be skipped. Controls whether resolvers run to compute features: - If true, all output features are recomputed by resolvers - If false, all output features are sampled from the offline store - If a list, features in the list are recomputed, others are sampled The environment in which to run resolvers. A unique name for the dataset. If provided, the dataset will be saved and can be retrieved later. Maximum number of samples to include in the result. If not specified, all samples are returned. Only query data observed after this timestamp. Accepts ISO 8601 format strings. Only query data observed before this timestamp. Accepts ISO 8601 format strings. Override the feature whose values are filtered against lower_bound and upper_bound. By default, the bounds filter against the FeatureTime feature of each output's namespace. Must reference a scalar or FeatureTime feature. The tags used to scope the resolvers. If specified, routes to the relevant branch. A globally unique ID for the query, used in logs and web interfaces. The name of the query. If provided, creates a named query or fills in missing parameters from a preexisting execution. If true, runs the query in separate Kubernetes pods. Useful for large datasets and long-running jobs. If true, stores the query output in the online store. If true, stores the query output in the offline store. If specified, splits the input across this many shards for parallel processing. Override resource requests (CPU, memory) for the offline query job. ### Response The response contains information about the submitted offline query job. A unique identifier for the offline query job. Use this to check the job status. If a dataset_name was provided, this is the ID of the created dataset. ### Check Offline Query Status Check the status of an offline query job. Offline queries run asynchronously, so you need to poll this endpoint to determine when the job is complete and the results are ready. ### Request The job ID returned from the offline query request (also called revision_id). The name of the dataset, if one was provided in the offline query request. The ID of the dataset. If true, returns results even if some errors occurred during execution. Default is false. If true, skips failed shards and returns results from successful shards only. Default is false. You must provide at least one of: job_id, dataset_name, or dataset_id. ### Response Whether the offline query job has completed. Poll this endpoint until this field is true. A list of short-lived, authenticated URLs to download the query results. These URLs point to data files in cloud storage (S3 or GCS) containing the feature values in a columnar format. Only populated when is_finished is true. Errors encountered during query execution, if any. Version number representing the format of the data. The client uses this to properly decode and load the query results into DataFrames. Current version is 1. Once is_finished is true, you can download the data from the provided URLs and load it into a DataFrame. The ChalkClient's Dataset object handles this automatically. # AI Router source: https://docs.chalk.ai/docs/ai-router ## Route LLM traffic to multiple providers through one OpenAI-compatible gateway. The Chalk AI Router is a managed, OpenAI-compatible LLM gateway hosted alongside your Chalk deployment. Point any OpenAI-compatible client at the router and it forwards your request to the right provider — OpenAI, Anthropic, or Google Gemini — behind a single API. Because the router sits in front of every provider, it is also where you centralize the things you do not want scattered across application code: API keys and budgets, rate limits, automatic fallback between models, and provider credentials. The router is hosted with the Chalk API and requires no installation or self-hosting. ### Endpoint The router exposes the standard OpenAI-compatible paths under this base URL: | Path | Purpose | | --------------------- | -------------------------------------------------- | | `/chat/completions` | Chat completions (streams via Server-Sent Events). | | `/embeddings` | Text embeddings. | | `/images/generations` | Image generation. | | `/models` | List the models available to your key. | If you are on a dedicated or self-hosted Chalk deployment, replace api.chalk.ai with your own API server host. You can find it in the apiServer field of chalk config --format json. Requests are routed to the calling environment's router, selected by the X-Chalk-Env-Id header described below. ### Authentication Every request carries an issued router API key as a bearer token, plus the environment to route to: | Header | Value | | ---------------- | --------------------------------------- | | `Authorization` | `Bearer ` | | `X-Chalk-Env-Id` | The environment to route the request to | Router API keys are issued and revoked from the dashboard — see API keys below. A key is shown only once when it is issued, so copy it immediately. ### Using the router The router speaks the OpenAI API, so any OpenAI-compatible client works with no code changes beyond the base URL, key, and environment header. ``` from openai import OpenAI client = OpenAI( base_url="https:///v1/router", api_key="", default_headers={"X-Chalk-Env-Id": ""}, ) resp = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "Hello"}], ) print(resp.choices[0].message.content) ``` The same request with curl: ``` curl https:///v1/router/chat/completions \ -H "Authorization: Bearer $CHALK_ROUTER_API_KEY" \ -H "X-Chalk-Env-Id: " \ -H "Content-Type: application/json" \ -d '{"model": "gpt-4o", "messages": [{"role": "user", "content": "Hello"}]}' ``` The model you request is resolved against the providers you have configured. A key's provider restriction, model allow-list, team, and daily token budget are all enforced on every request. ### Calling the router from Chalk Because the router is OpenAI-compatible, you can also call it from inside feature computation with chalk.functions. F.openai_complete issues a chat completion while a query runs. To route it through the AI Router rather than calling OpenAI directly, set api_server to your Chalk API host — or set the OPENAI_BASE_URL environment variable on the execution host and omit the argument. ``` import chalk.functions as F from chalk.features import _ from chalkdf import DataFrame df = DataFrame({"questions": ["Recommend some movies like High and Low by Akira Kurosawa"]}) ( df.with_columns( F.openai_complete( prompt=_.x, model="o4-mini", service_tier="flex", api_server="https://api.chalk.ai", # route through the AI Router ).alias("result") ) .run() .to_arrow() ) ``` F.openai_complete takes the following arguments: | Argument | Description | | -------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `prompt` | The prompt text to send to the model. | | `model` | The model to use. Defaults to `gpt-3.5-turbo` when omitted. | | `api_server` | Base URL of an OpenAI-compatible endpoint; `/chat/completions` is appended. Falls back to the `OPENAI_BASE_URL` env var, then to the default OpenAI endpoint. Point it at your Chalk API host to use the AI Router. | | `api_key` | API key for authentication. Falls back to the `OPENAI_API_KEY` env var, so the secret does not have to be threaded through feature data. | | `max_tokens` | Maximum number of tokens to generate. | | `temperature` | Sampling temperature between 0 and 2. | | `service_tier` | Optional OpenAI service tier — `"flex"` (cheaper, higher-latency), `"priority"`, or `"auto"`. `"flex"` is only supported on reasoning models (`o3`, `o4-mini`, `gpt-5`-class) and uses a longer request timeout; passing it with an unsupported model returns `null`. | It returns a struct with the completion text plus prompt_tokens, completion_tokens, total_tokens, model, finish_reason, and the upstream ratelimit_remaining_tokens / ratelimit_remaining_requests headers. ### Throttling with rate limits LLM calls are blocking and metered, so throttle them with the policy modifiers chained onto the expression. with_rate_limit caps how often the call may run across every expression that shares its key: ``` import chalk.functions as F from chalk.features import _ from chalkdf import DataFrame df = DataFrame({"x": ["Recommend some movies like Buzzard by Joel Potrykus"]}) ( df.with_columns( F.openai_complete(prompt=_.x, model="o4-mini", service_tier="flex") .with_rate_limit(rate=3, key="openai", per="minute") .alias("result") ) .run() .to_arrow() ) ``` The same modifier works on a feature defined in a @features class — chain .with_rate_limit(...) before selecting the .completion field. with_rate_limit takes: | Argument | Description | | ------------------ | ---------------------------------------------------------------------------------------------- | | `rate` | Number of calls allowed per window. | | `per` | Window length — `"second"` (default), `"minute"`, or `"hour"`. | | `key` | Bucket name; all expressions sharing a `key` draw from the same budget. | | `enforce_globally` | When `True`, enforce the limit across all workers rather than per-worker. Defaults to `False`. | These policy modifiers compose — chain with_concurrency, with_rate_limit, and with_retry on the same expression to bound in-flight calls, cap the call rate, and retry transient failures with backoff: ``` ( F.openai_complete(prompt=_.x, model="o4-mini", service_tier="flex") .with_concurrency(max_concurrent=4, key="my_api") .with_rate_limit(rate=100, key="my_api") .with_retry(max_retries=3, key="my_api") ) ``` Reusing one key ties the policies to the same logical resource, so every expression that calls "my_api" shares a single budget — here, at most 4 concurrent calls and 100 calls per second (per defaults to "second"), with each failed call retried up to 3 times. ### API keys Issue and revoke router keys from the AI Router → API keys tab of the Chalk dashboard. Each key can be scoped at issue time so that a leaked or over-eager client cannot do more than you intend: | Setting | Effect | | ---------------------- | ---------------------------------------------------------------------------- | | **Description** | A human-readable label for the key. | | **Team** | Associates the key with a [team](#teams), so team-scoped rate limits apply. | | **Provider** | Restricts the key to a single provider (for example, `openai`). | | **Model allow-list** | Restricts the key to specific models (for example, `gpt-4o`, `gpt-4o-mini`). | | **Daily token budget** | Caps the tokens the key may consume per day. | | **Labels** | Custom key/value metadata. | | **Cost tags** | Tags used to attribute spend for billing and reporting. | Each key tracks its total token usage, and revoking a key takes effect immediately. ### Teams Teams group API keys so you can manage and limit access by team rather than key by key. Create a team, then assign keys to it at issue time. Rate limits can be scoped to a team so the whole team shares a single budget. Deleting a team leaves its keys working but removes their team association. ### Rate limits Rate limit policies cap throughput and protect against runaway spend. Each policy has a limit type, an optional target, and a scope: - Limit type — tokens per minute, requests per minute, or concurrent requests. - Target (optional) — narrow the policy to a specific provider or model. Leave it unset to apply across all traffic. - Scope — apply the limit per individual key (token) or shared across a team. Policies can be enabled or disabled without deleting them. ### Fallback policies A fallback policy keeps requests succeeding when a primary model is unavailable. For a primary model you define an ordered list of fallbacks; if a request against the primary fails, the router retries the fallbacks in order. For example: ``` gpt-4o → [gpt-4o-mini, claude-3-5-sonnet] ``` If gpt-4o is unavailable, the router tries gpt-4o-mini, then claude-3-5-sonnet, before giving up. Edit fallback rules in the AI Router → Fallback tab. ### Providers Configure provider credentials in the AI Router → Settings tab. The router supports: | Provider | ID | | ------------- | ----------- | | OpenAI | `openai` | | Anthropic | `anthropic` | | Google Gemini | `gemini` | For each provider you set an API key and, optionally, a base URL override to point at a custom or OpenAI-compatible endpoint (for example, Gemini's OpenAI-compatible API). Credentials are applied at runtime — no redeploy is required — and each provider shows a connected or not-configured status. ### Playground The AI Router → Playground tab is an in-dashboard tester for the router. It authenticates with your Chalk session — no separate API key needed — and lets you exercise the router across three tabs: - Chat — send streaming chat completions with a configurable system prompt, temperature, and max tokens. - Embeddings — generate embeddings and inspect their dimensions and token counts. - Images — generate images with configurable size, quality, and count. The model picker is populated from the router's /models endpoint, so it reflects the providers you have configured. ### See also - Authentication — service credentials and RBAC for the Chalk API. - MCP Gateway — govern the external MCP servers your agents reach. - LLM Toolchain — building AI features and agents on Chalk. # MCP Server source: https://docs.chalk.ai/docs/mcp-server ## Connect AI agents to your Chalk deployment over the Model Context Protocol. Chalk hosts a Model Context Protocol (MCP) server so that AI agents — Claude, Cursor, your own LangChain/agent code, or any MCP-compatible client — can interact with your Chalk deployment directly. Through the MCP server an agent can run online queries, execute ChalkSQL, read feature and resolver definitions, search logs and traces, and inspect query errors, all scoped to the permissions of the credentials it presents. The MCP server is hosted alongside the Chalk API and requires no installation or self-hosting. ### Endpoint The server speaks the Streamable HTTP transport (MCP specification 2025-03-26). A single URL handles both the JSON-RPC POST requests and the Server-Sent Events (SSE) streams that the transport uses, so most clients only need this one URL. Despite the /sse suffix in the path, this is the modern Streamable HTTP transport rather than the deprecated HTTP+SSE transport — configure your client for a streamable-HTTP (or "HTTP") remote server, not a legacy SSE server. If you are on a dedicated or self-hosted Chalk deployment, replace api.chalk.ai with your own API server host. You can find it in the apiServer field of chalk config --format json. ### Authentication The MCP server authenticates with the same service credentials used everywhere else in Chalk. Provide your client_id and client_secret as HTTP headers on every request: | Header | Value | | ----------------------- | ------------------------------------ | | `X-Chalk-Client-Id` | Your service token's `client_id` | | `X-Chalk-Client-Secret` | Your service token's `client_secret` | The server exchanges these for a short-lived access token internally, so you do not need to run the OAuth client-credentials flow yourself. ### Getting your credentials Generate or look up a service token in the Settings → Service Tokens tab of the Chalk dashboard, then retrieve the values from the CLI with chalk config: ``` chalk config ``` ``` Name Value Description client_id token-392c737aa1e467e42e85ae3e8417a003 default token client_secret ts-6307f46a00a68436b0f955b82b7fb30075d default token environment btfxgaqqxbt7z ... api_server https://api.chalk.ai ... ``` To pull the values out programmatically, use a machine-readable format: ``` # JSON object with clientId / clientSecret / apiServer fields chalk config --format json # Shell export statements: CHALK_CLIENT_ID / CHALK_CLIENT_SECRET / ... chalk config --format shell ``` ### Bearer tokens and OAuth The server also accepts a standard Authorization: Bearer header. If a request arrives without credentials, the server responds with 401 and a WWW-Authenticate challenge pointing at Chalk's OAuth discovery endpoints (/.well-known/oauth-protected-resource), so MCP clients that implement the MCP authorization spec (including Dynamic Client Registration) can negotiate a token interactively. For headless and server-to-server use, the X-Chalk-Client-Id / X-Chalk-Client-Secret headers are the simplest option. ### Selecting an environment By default, requests run against the environment associated with your service token. To target a different environment, either: - send the X-Chalk-Env-Id: header, or - pass an environment_id argument to an individual tool call. Environment overrides are only honored for user (personal) credentials. Use the list_environments tool to discover the environment IDs available to your team. ### Connecting a client Most MCP clients accept a remote server URL plus a set of headers. Use the streamable-HTTP / remote-server configuration for your client and point it at the endpoint above. ### JSON configuration This is the format used by Cursor, Windsurf, and many other clients (mcp.json / mcpServers): ``` { "mcpServers": { "chalk": { "url": "https://api.chalk.ai/v1/mcp/sse", "headers": { "X-Chalk-Client-Id": "token-392c737aa1e467e42e85ae3e8417a003", "X-Chalk-Client-Secret": "ts-6307f46a00a68436b0f955b82b7fb30075d" } } } } ``` ### Claude Code Add the server from the command line with claude mcp add: ``` claude mcp add --transport http chalk https://api.chalk.ai/v1/mcp/sse \ --header "X-Chalk-Client-Id: $(chalk config --format json | jq -r .clientId)" \ --header "X-Chalk-Client-Secret: $(chalk config --format json | jq -r .clientSecret)" ``` ### Clients that only support stdio For a client that cannot connect to a remote server directly, bridge to it with mcp-remote: ``` { "mcpServers": { "chalk": { "command": "npx", "args": [ "mcp-remote", "https://api.chalk.ai/v1/mcp/sse", "--header", "X-Chalk-Client-Id: ${CHALK_CLIENT_ID}", "--header", "X-Chalk-Client-Secret: ${CHALK_CLIENT_SECRET}" ], "env": { "CHALK_CLIENT_ID": "token-392c737aa1e467e42e85ae3e8417a003", "CHALK_CLIENT_SECRET": "ts-6307f46a00a68436b0f955b82b7fb30075d" } } } } ``` ### Verifying the connection with curl You can confirm your credentials are accepted by issuing an initialize request directly: ``` curl -sS https://api.chalk.ai/v1/mcp/sse \ -H "X-Chalk-Client-Id: $(chalk config --format json | jq -r .clientId)" \ -H "X-Chalk-Client-Secret: $(chalk config --format json | jq -r .clientSecret)" \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"curl","version":"0"}}}' ``` A 401 with a WWW-Authenticate header means the credentials were missing or rejected. ### Available tools The server exposes a set of tools for querying and operating your Chalk deployment. The exact set evolves over time, and the tools a given client can call depend on the permissions granted to its credentials — for example, run_online_query requires the query.online permission, and the observability tools require monitoring.read. Tools your token is not authorized for will return a permission error. ### Queries and SQL | Tool | Description | | --------------------- | ----------------------------------------------------------------------------------- | | `run_online_query` | Run an online query for one or more features. | | `execute_sql_query` | Execute a [ChalkSQL](/docs/chalksql/what-is-chalk-sql) query, with preview support. | | `get_offline_queries` | List offline queries (batch jobs). | | `get_offline_query` | Fetch metadata for a single offline query. | ### Metadata and discovery | Tool | Description | | --------------------------------------------------- | ---------------------------------------------------------- | | `list_features` | List the features defined in the environment. | | `get_feature_definition` | Read a feature's definition and metadata. | | `list_resolvers` | List resolvers. | | `get_resolver_definition` | Read a resolver's definition and metadata. | | `list_functions` / `get_function` / `call_function` | List, inspect, and invoke external functions. | | `list_branches` | List the deployment's branches. | | `list_environments` | List environments available to your team (with their IDs). | | `search_deployment_source` | Search the deployed source code. | | `search_docs` | Search the Chalk documentation. | ### Observability and debugging | Tool | Description | | ----------------------- | ------------------------------------------------------------------ | | `get_query_errors` | List recorded query errors, with filtering and pagination. | | `get_query_plan_json` | Retrieve a query plan as JSON. | | `get_perf_summary_json` | Retrieve a query performance summary as JSON. | | `search_logs` | Search application logs for the environment. | | `search_traces` | Search distributed traces and their spans. | | `search_kube_events` | Search Kubernetes cluster events (scheduling, OOM kills, crashes). | Additional infrastructure and sandbox tools (Kubernetes deployment/node inspection, sandboxed Python execution, and similar) are available to credentials with the appropriate elevated permissions. ### See also - Authentication — service credentials and RBAC for the Chalk API. - Development with LLMs — prompts for writing Chalk code with AI assistants. - ChalkSQL — the SQL dialect available through execute_sql_query. # Offline Store source: https://docs.chalk.ai/docs/offline-stores ## Historical feature storage and retrieval. The Chalk offline store persists all computed feature values — enabling monitoring and dataset generation for model training. It is backed by a data warehouse such as BigQuery, Snowflake, or Iceberg. While the online store — backed by low-latency stores like Redis or DynamoDB — caches the latest value of each feature for fast serving, the offline store retains a deep historical record of feature values, timestamped and keyed by entity. The offline store is typically implemented using BigQuery, Snowflake, Iceberg, or other data warehouses. ### How data flows into the offline store Feature values can be written to the offline store from several sources: - Online queries — freshly computed features from online queries are automatically persisted to the offline store. This means your offline store passively accumulates a historical record of every feature value served in production. - Offline queries - offline queries can write to the offline store if specified. - Scheduled resolver runs — scheduled or triggered resolver runs execute in batch across your upstream data sources and write all computed values to the offline store. - Backfills — you can explicitly ingest historical feature data from external sources using batch backfilling. ### How features are stored: Fully Qualified Names Each feature in Chalk has a Fully Qualified Name (FQN) that uniquely identifies it. An FQN is composed of: ``` . ``` The namespace comes from the @features class name (stripped of a trailing "Features" suffix if present), and the feature name comes from the Python attribute name — unless overridden via feature(name=...) or @features(name=...). For example: ``` @features class User: id: int fraud_score: float ``` | Python attribute | FQN | | ------------------------- | -------------------------- | | `UserExample.id` | `user_example.id` | | `UserExample.fraud_score` | `user_example.fraud_score` | Historical feature value tables in the offline store are named after a hash of the feature's namespace, FQN, and internal version. More information on the historical feature value table schema can be found in Chalk catalog components. ### Internal versioning and type safety In addition to the FQN, Chalk tracks an internal version for every feature when naming its offline store table. This internal version is managed automatically and is separate from the explicit user-defined versions you can set with feature(version=...). The internal version increments whenever you change the type of a feature. For example, changing fraud_score from float to int will cause Chalk to write new values into a new table, rather than mixing them with the existing one. This guarantees that the offline store never serves values of incompatible types for the same feature. In practice: - Type changes are safe — old data is preserved under the previous internal version and is no longer served, while new values accumulate in the new table. - No manual migration is needed — Chalk handles the versioning automatically on deployment. - History resets on type change — if you need to preserve historical values after a type change, backfill the new table explicitly. ### Querying the offline store There are a few ways to query the offline store: Offline queries — use ChalkClient.offline_query() from the Python client to retrieve historical feature values, build training datasets, and run batch inference. See Offline Queries for the full API reference, including point-in-time lookups, spine SQL input, timebounds, and large query execution. SQL Explorer — run SQL directly against historical feature value tables from the Chalk dashboard. Navigate to the SQL Explorer tab to query the offline store without writing any code. See Chalk SQL for more detail. ### Offline store options Chalk supports several offline store backends. In general, choose the store you already use in your data platform — Chalk handles routing and FQN-based isolation regardless of backend. | Store | Format | Cloud | Setup guide | | --------------------- | --------------------------- | ---------- | ------------------------------------------------------------------------------------------------- | | Google BigQuery | Columnar (Bigtable-based) | GCP | [BigQuery](/docs/bigquery) | | Amazon Redshift | Columnar (Parquet) | AWS | — | | Snowflake | Columnar (Micro-partitions) | GCP or AWS | [Snowflake (AWS)](/docs/snowflake-deployment) · [Snowflake (GCP)](/docs/snowflake-gcp-deployment) | | Databricks Delta Lake | Columnar (Parquet) | Any | [Databricks](/docs/databricks_offline_store) | | Iceberg | Columnar (Parquet) | Any | [Iceberg](/docs/iceberg-deployment) | For GCP deployments, BigQuery is generally recommended for its performance with analytical queries. For AWS deployments, Redshift or Snowflake are common choices depending on your existing data platform. # Getting Started source: https://docs.chalk.ai/docs/getting-started ## Your first time with Chalk ### Introduction Chalk's platform is a powerful tool that allows data engineers and data scientists to collaborate efficiently and effectively. In this project we'll create features related to Usersand their credit scores. We'll combine data from an API and a database to help us decide whether we should issue new loans. You will learn how to create a new Chalk project, create features and resolvers, deploy them to the Chalk environment, and query the environment using the Chalk CLI and a Jupyter Notebook. This tutorial will take about 45 minutes to complete end-to-end. ### Installing the Required Software - If you don't have a favorite IDE, install VS Code by visiting the download site and running the appropriate installer. - You'll also want to install the code CLI command. - If you haven't already, install Python. You can do this through Homebrew as follows: ``` # Install Homebrew $ /bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)" # Use Homebrew to install python3 $ brew install python ``` If you're having trouble installing Python, follow the instructions in Python's docs. - Install the Chalk CLI by running ``` $ curl -s -L https://api.chalk.ai/install.sh | sh ``` ### Creating your Chalk project Now that all of your software is installed, let's create a Chalk project where you will add your features and resolvers. - Create a folder and call it chalk-tutorial. - Run chalk init from inside the chalk-tutorial to initialize your project files. This will create two files for you: chalk.yaml and .chalkignore. The first file, chalk.yaml, contains configuration information about your project. The second file, .chalkignore, tells your Chalk CLI tool which files to ignore when deploying to your Chalk environment. You can edit this file and use it just like a .gitignore file. ``` $ mkdir chalk-tutorial $ cd chalk-tutorial $ chalk init Created project config file chalk.yaml Created .chalkignore file ``` - Now edit the chalk.yaml file and set the project field to the name of your Chalk project. This might be chalk-tutorial, but check by visiting the Projects page. - Login by typing chalk login. If you're using a dedicated environment, make sure you use the --api-host flag. Type y when prompted and login in the browser. Now you're ready to deploy code and query the environment! ### Create a virtual environment We're going to create a virtual environment so that you can download and use Python libraries in your Chalk project. - Run the following command from inside your Chalk project to create and activate your virtual environment. When you want to deactivate your virtual environment, type deactivate. ``` $ python3 -m venv chalk-tutorial-venv $ source chalk-tutorial-venv/bin/activate ``` - Your requirements.txt file tells your virtual environment (and Chalk) which libraries to install. We'll use a few libraries for this tutorial, so add the following to the requirements.txt. Feel free to add more libraries later. ``` chalkpy pydantic requests ``` - Install the requirements. ``` $ pip3 install -r requirements.txt ``` - Check that the libraries are installed by importing requests in Python. If there's no error message, it worked. Exit by typing exit() and hitting enter. ``` $ python3 >>> import requests >>> exit() ``` ### Use Curl and Jupyter to call APIs. Now that all the software is installed and ready to use, you're ready to start integrating an API. - First, lets test the API we're going to use. Run this command to hit the API from your command line. Here we're asking "give me the credit score for user 12" and the API gives us back a JSON object with a credit score. ``` $ curl https://credit-report.chalk.dev/rutter_score/12 {"score":98} ``` Now you should try to hit this API from Python using a Jupyter Notebook. - Open VS Code, and create a new file File -> New File. Select Jupyter Notebook. - Jupyter will probably ask you which interpreter to use. If it does, select the virtual environment you created before (chalk-tutorial-venv). - If not, you can select it manually. Select Jupyter virtual environment - Finally, let's query the API from Jupyter. Enter the following code and run it with shift+enter. ``` import requests requests.get( "https://credit-report.chalk.dev/rutter_score/12" ).json()["score"] ``` If you see 98, congratulations! You're all setup and ready to start developing your Chalk project. ### Building a Chalk project With all the pre-requisites completed, let's jump right into building your Chalk project. ### Creating your first Chalk feature We'll create a feature called User with a "Rutter score" property. The resolver will call the "Rutter" API we tested earlier, passing the User ID, and return the score. After you deploy the feature, you will test it in Jupyter. - Create a credit.py file and copy in the feature and resolver definitions below: ``` import requests from chalk.features import online, features, Features @features class User: id: str credit_score: int @online def get_credit_score( id: User.id, ) -> Features[User.credit_score]: return requests.get( f"https://credit-report.chalk.dev/rutter_score/{id}" ).json()["score"] ``` - Deploy your code. Pass the --branch flag with a name so that you don't affect the "actual" environment ``` chalk apply --branch test ``` - Now lets query for your feature from your Jupyter Notebook. Enter and execute the following code in your notebook. ``` from credit import User from chalk.client import ChalkClient ChalkClient().query( input={User.id: 'u_F6zY0tE4w8'}, output=[User.credit_score], branch="test" ) ``` You can also query for this feature in your terminal: ``` $ chalk query --in user.id=u_F6zY0tE4w8 \ --out user.credit_score \ --branch test ``` Congratulations! You've written your first Chalk feature, connected it to an external API, and queried it from a Jupyter Notebook. This is a big achievement! ### Adding a database to Chalk We have a database containing more information about users, and it would be good to combine that information together on our user feature. - If you have psql or similar tool installed, you can preview the database from the command line. When prompted, the password is postgres. ``` psql -U postgres -p 5432 -h 35.188.7.252 -d postgres postgres=> select * from users limit 10; ``` Notice that we know name, surname, email, birthday, and is_fraud, a flag that tells us whether we know this user has committed fraud in the past. - To use this in your Chalk project, we'll add the database through the UI. Navigate to your environment, and find "Data Sources" on the left hand navigation. - If you see the database already, great, you can skip the next step of adding the database. Otherwise, continue. - Click Add a Data Source and choose PostgreSQL. - Fill out the fields as shown here. The password is postgres. Postgres ### Write a resolver that resolves a database - Add the fields from the database to the User feature class you created before. The properties name, surname, and email are strings. Birthday is a date, and is_fraud is a boolean. (Hint: you may have to add from datetime import date to your credit.py file) - Add the datasource to your Python context. Add the below code to your credit.py file. This tells Python about your datasource. ``` from chalk.sql import PostgreSQLSource user_pg = PostgreSQLSource(name="User_PG") ``` - You can use a SQL File Resolver to pull this information. Create a new file called users.chalk.sql and copy the below contents: ``` -- type: online -- resolves: user -- source: postgres -- count: 1 select name, surname, email, birthday, is_fraud from users where id=${user.id} ``` - After deploying (remember to use --branch with the name you specified, for example "test"), query your feature. Supplying User as the output tells Chalk to give you back ALL the features of User. ``` res = client.query( input={User.id: 'u_F6zY0tE4w8'}, output=[User], branch="test" ) ``` You can also query for this feature in your terminal: ``` $ chalk query --in user.id=u_F6zY0tE4w8 \ --out user \ --deployment test ```