Chronon

https://zipline.ai/docs
Admin
Chronon is an open source end-to-end feature platform that allows Machine Learning teams to easily...
Tokens:41,088
Snippets:419
Trust Score:6.1
Update:4 months ago
Show doc for...
Tokens:
Raw
### Download and Setup Spark

Source: https://zipline.ai/docs/Kaggle_Outbrain

Download, extract, and configure Spark for local use. Ensure a JDK is already installed. Set SPARK_SUBMIT_PATH and SPARK_LOCAL_IP environment variables.

```bash
cd ~
curl -O https://archive.apache.org/dist/spark/spark-3.2.4/spark-3.2.4-bin-hadoop3.2.tgz
tar xf spark-3.2.4-bin-hadoop3.2.tgz
export SPARK_SUBMIT_PATH=/Users/$USER/spark-3.2.4-bin-hadoop3.2/bin/spark-submit
export SPARK_LOCAL_IP="127.0.0.1"
```

--------------------------------

### Download and Run Docker Compose

Source: https://zipline.ai/docs/getting_started/Local_Sandbox

Download the docker-compose.yml file and start the Chronon local environment. Ensure Docker is installed and running.

```bash
curl -o docker-compose.yml https://chronon.ai/docker-compose.yml
docker-compose up
```

--------------------------------

### StagingQuery with SQL Join

Source: https://zipline.ai/docs/authoring_features/StagingQuery

Example of a StagingQuery used to join 'fct_purchases' with 'dim_users' to prepare data for a join operation. It includes SQL query, UDF setup, and metadata dependencies.

```python
v1 = StagingQuery(
    query="""
        SELECT
            a.id_user,
            a.ts,
            a.ds,
            b.user_type,
            b.user_country,
            EMAIL_PARSE(b.email) as parsed_email,
        FROM
            data.fct_purchases a
        JOIN
            data.dim_users b
        ON
            a.id_user=b.id
        AND
            a.ds = b.ds
        WHERE
            a.ds between '{{ start_date }}' AND '{{ end_date }}'
        AND
            b.ds between '{{ start_date }}' AND '{{ end_date }}'
    """,
    setups=[
        "ADD JAR s3://path/to/your/jar",
        "CREATE TEMPORARY FUNCTION EMAIL_PARSE AS 'com.you_company.udf.your_team.YourUdfClass'",
    ],
    metaData=MetaData(
        dependencies=["data.users", "data.purchases"], # Used by airflow to setup partition sensors
    )
)
```

--------------------------------

### Setup Chronon Repository

Source: https://zipline.ai/docs/Kaggle_Outbrain

Clone the Chronon repository and set the PYTHONPATH environment variable to include the Chronon API directories. This is a one-time setup step.

```bash
cd ~/repos
git clone git@github.com:zipline/chronon.git
export PYTHONPATH=/Users/$USER/repos/chronon/api/python/:/Users/$USER/repos/chronon/api/python/test/sample/:$PYTHONPATH
```

--------------------------------

### Install Zipline AI CLI

Source: https://zipline.ai/docs/getting_started/Quickstart

Install the Zipline AI command-line interface. This is the first step to setting up the AI agent.

```bash
pip install zipline-ai
```

--------------------------------

### Java Client Fetch Join Example

Source: https://zipline.ai/docs/getting_started/Local_Sandbox

Illustrative example of using the Java client to fetch join data. This code is not runnable in the docker environment.

```java
Map<String, String> keyMap = new HashMap<>();
keyMap.put("user_id", "123");
Fetcher.fetch_join(new Request("quickstart/training_set_v1", keyMap))
```

--------------------------------

### Sample Response Format

Source: https://zipline.ai/docs/getting_started/Local_Sandbox

Example of a sample response format for fetched data.

```json
'{"purchase_price_avg_3d":14.3241, "purchase_price_avg_14d":11.89352, ...}'
```

--------------------------------

### Placeholder Syntax Example

Source: https://zipline.ai/docs/authoring_features/SecretsManagement

Demonstrates the basic syntax for using placeholders in configuration values. Placeholders are enclosed in curly braces and must follow specific naming conventions.

```json
"spark.chronon.jdbc.url": "jdbc://{DB_HOST}:{DB_PORT}/mydb"
```

--------------------------------

### Chaining Feature API Example

Source: https://zipline.ai/docs/authoring_features/ChainingFeatures

Demonstrates how to define an upstream join and a downstream groupby to compute the average price of listings viewed in the last 14 days per user. This example showcases using the output of a parent join as a source for a new groupby.

```python
# Chaining Feature API example

# Upstream Join. Regular chronon Join for last prices of listings
parent_join = Join(
  left=rating_driver,
  right_parts=[
    JoinPart(group_by=listing_price_last),
  ],
  online=True,
  realtime=True,
  output_namespace="chronon_test",
  team_override='ml_infra',
)

listing_price_last = GroupBy(
  name="chronon_test.join_right_part",
  sources=price_mutations,
  keys=["listing"],
  aggregations=[
    Aggregation(input_column="price", operation=Operation.LAST),
  ],
  output_namespace="chronon_test",
  online=True,
  team_override='ml_infra',
  accuracy=Accuracy.TEMPORAL,
)

# Downstream Groupby for avg price of listings viewed in last 14 days per user
# New API in bold
enriched_listings = Join(
  left = source_table,
  right_parts = [
    JoinPart(
      group_by = GroupBy(
        sources = JoinSource(join = parent_join, query=Query(...)),
        keys = ["user"],
        aggregations = [
          Aggregation(
            operation = Operation.LAST_K(100),
            input_column = "price_last",
            window = [Window(14, TimeUnit.DAYS)]
          )
        ],
        accuracy = Accuracy.TEMPORAL
      ))
  ],
  derivations = [
    Derivation(
      name = "avg_listing_view_price_last_14d",
      expression = "avg(enriched_views_price_last_last_100)"
    )
  ]
)

```

--------------------------------

### Defining a REST Endpoint with Environment Variables

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Configure a RestEndpoint using environment variables for dynamic URL construction and authentication. This example shows a GET request to a user service using a bearer token.

```python
endpoint=RestEndpoint(
    url="https://${env.API_HOST}/v1/users/{user_id}",
 
    auth=AuthConfig(
        auth_type=AuthType.BEARER_TOKEN,
        token="${env.USER_SERVICE_TOKEN}"
    ),
 
    headers={
        "X-Environment": "${env.ENVIRONMENT}"
    }
)
```

--------------------------------

### Example: Fetching and Serving Tiled Features

Source: https://zipline.ai/docs/Tiled_Architecture

Demonstrates how Chronon fetches and merges pre-aggregated tiles from a KV store to serve feature values. Assumes 1-hour tiles for 3-hour windows.

```text
[0:00, 1:00) -> [2, "B"]
[1:00, 2:00) -> [9, "A"]
[2:00, 3:00) -> [3, "C"]
```

--------------------------------

### Test External Source Configuration

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Demonstrates how to use the Chronon CLI to validate the configuration of an external REST source, specifically for the user service in this example.

```bash
# Use Chronon CLI to validate endpoint config
chronon test-external-source joins/fraud/features.py:user_service
```

--------------------------------

### StagingQuery with BigQuery Engine

Source: https://zipline.ai/docs/authoring_features/StagingQuery

Example of configuring a StagingQuery to use Google BigQuery as its compute engine. Requires importing EngineType.

```python
from ai.chronon.types import EngineType, StagingQuery

v1 = StagingQuery(
    query="SELECT * FROM my_dataset.my_table WHERE ds BETWEEN '{{ start_date }}' AND '{{ end_date }}'",
    engine_type=EngineType.BIGQUERY,
    metaData=MetaData(
        dependencies=["my_dataset.my_table"],
    )
)
```

--------------------------------

### Batch EventSource Example

Source: https://zipline.ai/docs/authoring_features/Source

Configure a Batch EventSource by omitting the topic from an EventSource. This is suitable for daily batch computations using data from a specified table.

```python
source = Source(
    events=EventSource(
        table="data.returns",
        topic=None, # This makes it a batch source
        query=Query(
            selects=select("user_id","refund_amt"),
        )
    ))

```

--------------------------------

### Streaming EntitySource Example

Source: https://zipline.ai/docs/authoring_features/Source

Implement a Streaming EntitySource for real-time entity data. This requires a snapshot table, a mutation table, and a mutation topic for change data capture.

```python
user_activity = Source(entities=EntitySource(
  snapshotTable="db_snapshots.users",
  mutationTable="db_mutations.users",
  mutationTopic="events.users_mutations",
  query=Query(
            selects=select("user_id","account_created_ds","email_verified"), # Select the fields we care about
        )
)

```

--------------------------------

### Initialize Zipline AI Agent for Specific Agent

Source: https://zipline.ai/docs/getting_started/Quickstart

Initialize the Zipline AI agent for a specific coding assistant without a prompt. Use this for direct setup.

```bash
zipline init-agent --agent claude    # Claude Code
```

```bash
zipline init-agent --agent codex     # Codex
```

```bash
zipline init-agent --agent cursor    # Cursor
```

```bash
zipline init-agent --agent gemini    # Gemini CLI
```

```bash
zipline init-agent --agent windsurf  # Windsurf
```

```bash
zipline init-agent --agent copilot   # GitHub Copilot
```

--------------------------------

### HTTP Request with Bearer Token

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

An example of an HTTP GET request demonstrating the Authorization header with a Bearer token. This is how the client would typically authenticate with the API.

```http
GET /resource/123 HTTP/1.1
Host: api.company.com
Authorization: Bearer eyJhbGciOiJIUzI1NiIs...
```

--------------------------------

### Example: Unique Top K for Structs

Source: https://zipline.ai/docs/authoring_features/GroupBy

Illustrates the unique_top_k aggregation for struct types, requiring 'sort_key' for ordering and 'unique_id' for deduplication.

```python
unique_top_k(k=3) on records with {sort_key: "score_95", unique_id: 123}, {sort_key: "score_87", unique_id: 456}
```

--------------------------------

### Streaming EventSource Example

Source: https://zipline.ai/docs/authoring_features/Source

Use a Streaming EventSource for real-time data ingestion from both historical tables and live event streams. It requires specifying a table, a topic, and a query with a time column.

```python
source = Source(
    events=EventSource(
        table="data.returns", # This points to the log table with historical return events
        topic="events.returns", # Streaming event
        query=Query(
            selects=select("user_id","refund_amt"), # Select the fields we care about
            time_column="ts") # The event time
    ))

```

--------------------------------

### Initialize Zipline AI Agent

Source: https://zipline.ai/docs/getting_started/Quickstart

Initialize the Zipline AI agent for your coding assistant. This command prompts you to select an agent and installs the necessary context files.

```bash
zipline init-agent
```

--------------------------------

### Example Variable Naming for Similar GroupBys

Source: https://zipline.ai/docs/authoring_features/NamingAndVersioning

Use descriptive variable names when defining multiple similar entities within the same file to differentiate them. This example shows two GroupBys with slight differences in their definitions.

```python
no_bots_v0 = GroupBy(
    ...
    # Includes a filter clause on the source
)

with_bots_v0 = GroupBy(
    ...
    # No filter clause
)
```

--------------------------------

### Sample GraphQL Response

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

An example of a JSON response structure from the GraphQL product catalog API. This illustrates the data format that the response_mapping will parse.

```json
{
  "data": {
    "product": {
      "name": "Wireless Headphones",
      "category": {
        "name": "Electronics"
      },
      "pricing": {
        "current": 129.99
      },
      "inventory": {
        "available": true
      },
      "reviews": {
        "averageRating": 4.5
      }
    }
  }
}
```

--------------------------------

### Batch EntitySource Example

Source: https://zipline.ai/docs/authoring_features/Source

Define a Batch EntitySource using only a snapshot table. This source type is used for batch processing of entity data based on daily snapshots.

```python
source = Source(
    entities=EntitySource(
        snapshotTable="data.users", # This points to a table that contains daily snapshots of the entire product catalog
        query=Query(
            selects=select("user_id","account_created_ds","email_verified"), # Select the fields we care about
        )
    ))

```

--------------------------------

### Handler Registration Failure Example

Source: https://zipline.ai/docs/authoring_features/ExternalSource

Illustrates an IllegalArgumentException that occurs when a requested external source handler is not registered with the system.

```text
IllegalArgumentException: user_service is not registered among handlers: [merchant_service, inventory_service]
```

--------------------------------

### Realtime Event GroupBy with Multiple Aggregations

Source: https://zipline.ai/docs/authoring_features/GroupBy

Example demonstrating a GroupBy configuration for real-time event data. It aggregates 'refund_amt' using SUM, COUNT, AVERAGE, and LAST_K over specified window sizes.

```python
source = Source(
    events=EventSource(
        table="data.returns", # This points to the log table with historical return events
        topic="events.returns",
        query=Query(
            selects=select("user_id","refund_amt"), # Select the fields we care about
            time_column="ts") # The event time
    ))
 
window_sizes = ["3d", "14d", "30d"] # Define some window sizes to use below
 
v1 = GroupBy(
    sources=[source],
    keys=["user_id"], # We are aggregating by user
    online=True,
    aggregations=[
        Aggregation(
            input_column="refund_amt",
            operation=Operation.SUM,
            windows=window_sizes
        ), # The sum of purchases prices in various windows
        Aggregation(
            input_column="refund_amt",
            operation=Operation.COUNT,
            windows=window_sizes
        ), # The count of purchases in various windows
        Aggregation(
            input_column="refund_amt",
            operation=Operation.AVERAGE,
            windows=window_sizes
        ),
        Aggregation(
            input_column="refund_amt",
            operation=Operation.LAST_K(2),
        ),
    ],
)
```

--------------------------------

### Define a Simple GET REST External Source

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Configure a REST ExternalSource for a simple GET request. This includes defining the key and value schemas, the endpoint URL, HTTP method, static headers, authentication using environment variables, response mapping via JSONPath, request timeout, and retry configuration.

```python
from ai.chronon.api import RestExternalSource, RestEndpoint, AuthConfig, AuthType, MetaData
from ai.chronon.types import StructType, StructField, StringType, LongType, IntType, BooleanType, DoubleType

user_service = RestExternalSource(
    metadata=MetaData(name="user_service", team="fraud"),

    # Define keys needed to make the request
    key_schema=StructType("keys", [
        StructField("user_id", LongType)
    ]),

    # Define values returned from the API
    value_schema=StructType("values", [
        StructField("country", StringType),
        StructField("phone_verified", BooleanType),
        StructField("account_age_days", IntType),
        StructField("risk_score", DoubleType)
    ]),

    # Configure the REST endpoint
    endpoint=RestEndpoint(
        # URL with placeholder for user_id from key_schema
        url="https://user-service.company.com/v2/users/{user_id}",

        method=HttpMethod.GET,

        # Static headers
        headers={
            "Accept": "application/json",
            "X-Client-Id": "chronon-features"
        },

        # Authentication - token read from environment variable
        auth=AuthConfig(
            auth_type=AuthType.BEARER_TOKEN,
            token="${env.USER_SERVICE_TOKEN}"
        ),

        # Map JSON response fields to value_schema fields using JSONPath
        response_mapping={
            "country": "$.user.location.country_code",
            "phone_verified": "$.user.verification.phone_verified",
            "account_age_days": "$.user.account.age_in_days",
            "risk_score": "$.user.risk_assessment.score"
        },

        # Request timeout in milliseconds
        timeout_ms=3000,

        # Retry configuration
        retry=RetryConfig(
            max_retries=2,
            initial_backoff_ms=100,
            max_backoff_ms=1000
        )
    )
)

# Use in a Join like any other ExternalSource
my_join = Join(
    metadata=MetaData(name="fraud_model", team="fraud"),
    left=Source.events(...),
    right_parts=[...],
    online_external_parts=[
        ExternalPart(user_service, prefix="usr")
    ]
)
```

--------------------------------

### Third-Party Enrichment REST Source (OAuth2 with Batching)

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Set up a RestExternalSource for a third-party enrichment service using OAuth2 authentication and batching. This example demonstrates configuring batching parameters and rate limiting.

```python
enrichment_service = RestExternalSource(
    metadata=MetaData(name="enrichment", team="fraud"),
    key_schema=StructType("keys", [StructField("email_hash", StringType)]),
    value_schema=StructType("values", [
        StructField("income_bucket", StringType),
        StructField("age_range", StringType)
    ]),
    endpoint=RestEndpoint(
        url="https://api.thirdparty.com/v3/batch",
        method=HttpMethod.POST,
        auth=AuthConfig(
            auth_type=AuthType.OAUTH2,
            client_id="${env.ENRICHMENT_CLIENT_ID}",
            client_secret="${env.ENRICHMENT_SECRET}",
            token_url="https://auth.thirdparty.com/token",
            scopes=["enrichment.read"]
        ),
        batch_config=BatchConfig(
            enabled=True,
            max_batch_size=50,
            batch_key_path="$.email_hashes",
            batch_result_path="$.results[* P*]",
            result_key_field="email_hash"
        ),
        response_mapping={
            "income_bucket": "$.demographics.income_bucket",
            "age_range": "$.demographics.age_range"
        },
        timeout_ms=5000,
        rate_limit=RateLimitConfig(requests_per_second=50)
    )
)
```

--------------------------------

### SCD2 Aggregation with Temporal Filtering

Source: https://zipline.ai/docs/authoring_features/SCD2

This example demonstrates how to aggregate data by filtering active rows at a given timestamp and folding them, specifically for SUM operations on amounts.

```sql
aggregate(
  filter(
    map_values(__chronon_scd2_row_last_by___chronon_scd2_row_id),
    x -> x.__chronon_scd2_valid_from <= ts AND ts < x.__chronon_scd2_valid_to
  ),
  named_struct('sum', cast(0 as double), 'cnt', cast(0 as bigint)),
  (acc, x) -> named_struct(
    'sum', acc.sum + IF(x.amount IS NULL, cast(0 as double), cast(x.amount as double)),
    'cnt', acc.cnt + IF(x.amount IS NULL, cast(0 as bigint), cast(1 as bigint))
  ),
  acc -> IF(acc.cnt = 0, cast(NULL as double), acc.sum)
)
```

--------------------------------

### ContextualSource Example

Source: https://zipline.ai/docs/authoring_features/ExternalSource

Define and use ContextualSource to pass request context like identifiers and experiment metadata to features without external calls. Specify fields and row_ids for join matching.

```python
from ai.chronon.api import ContextualSource, ExternalPart

# Define contextual fields
contextual_source = ContextualSource(
    fields=[
        ("event_id", StringType),
        ("session_id", StringType),
        ("experiment_variant", StringType)
    ]
)

# Add to Join
my_join = Join(
    left=...,
    right_parts=[...],
    online_external_parts=[
        ExternalPart(contextual_source)
    ],
    # These row_ids are used for bootstrap join matching
    row_ids=["event_id"]
)
```

--------------------------------

### Zipline Hub Configuration Example

Source: https://zipline.ai/docs/authoring_features/SecretsManagement

Configure Zipline Hub to resolve a vault URI and interpolate the secret into Spark configuration. The vault URI is defined in 'env', and the resolved secret is used in 'spark.sql.catalog.spark_catalog.credential'.

```python
azure = Team(
    ...
    node=NodeProperties(
        env={
            # Vault URI → resolves to OC_CREDENTIAL env var
            "OC_CREDENTIAL_VAULT_URI": "https://dev-zipline-secrets.vault.azure.net/secrets/oc-catalog-credential/9dfce586bacb4c67b68514a9bdfb7a52",
            # Other non-secret env vars
            "SNOWFLAKE_JDBC_URL": "jdbc:snowflake://VEJLULX-AZURE.snowflakecomputing.com/?user=demo_batch_service&db=Demo&schema=public&warehouse=demo_wh",
        },
    ),
    conf=ConfigProperties(
        common={
            **OpenCatalogConfiguration({
                "spark.sql.catalog.spark_catalog.uri": "https://vejlulx-azure-oc.snowflakecomputing.com/polaris/api/catalog",
                # {OC_CREDENTIAL} is replaced with the resolved secret at submission time
                "spark.sql.catalog.spark_catalog.credential": "{OC_CREDENTIAL}",
                "spark.sql.catalog.spark_catalog.warehouse": "demo-v2",
            }),
        },
    ),
)

```

--------------------------------

### Example: Unique Top K for Strings

Source: https://zipline.ai/docs/authoring_features/GroupBy

Demonstrates the usage of unique_top_k aggregation for string types. It returns the top k unique elements based on their values, deduplicating inputs.

```python
unique_top_k(k=5) on ["apple", "banana", "apple", "cherry", "banana", "date"]
```

--------------------------------

### Start Shell Session in Docker Container

Source: https://zipline.ai/docs/getting_started/Local_Sandbox

Access the Chronon Docker container's shell to begin development. This command opens an interactive bash session inside the running container.

```bash
docker-compose exec main bash
```

--------------------------------

### Batch Entity GroupBy for Feature Extraction

Source: https://zipline.ai/docs/authoring_features/GroupBy

This example shows a Batch Entity GroupBy where the primary key of the GroupBy matches the source table's primary key. It extracts user fields as features without performing aggregations or using windows.

```python
"""
The primary key for this GroupBy is the same as the primary key of the source table. Therefore,
it doesn't perform any aggregation, but just extracts user fields as features.
"""
 
source = Source(
    entities=EntitySource(
        snapshotTable="data.users", # This points to a table that contains daily snapshots of the entire product catalog
        query=Query(
            selects=select("user_id","account_created_ds","email_verified"), # Select the fields we care about
        )
    ))
 
v1 = GroupBy(
    sources=[source],
    keys=["user_id"], # Primary key is the same as the primary key for the source table
    aggregations=None, # In this case, there are no aggregations or windows to define
    online=True,
)

```

--------------------------------

### Azure Key Vault URI Example

Source: https://zipline.ai/docs/authoring_features/SecretsManagement

Set the Azure Key Vault URI as an environment variable before job submission. The Chronon engine will resolve this URI to fetch the secret at driver startup.

```bash
export SNOWFLAKE_PRIVATE_KEY_VAULT_URI="https://demo-service-writer-pkey.vault.azure.net/secrets/snowflake-private-key"
```

--------------------------------

### Navigate to Chronon Sample Directory

Source: https://zipline.ai/docs/Kaggle_Outbrain

Switch to the sample directory within the Chronon API Python project. This is the primary location for development iterations.

```bash
cd api/python/test/sample/
```

--------------------------------

### User Service REST Source (GET Request)

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Define a RestExternalSource for a user service that makes GET requests. It includes schema definitions, authentication via bearer token, and response mapping for extracted fields.

```python
user_service = RestExternalSource(
    metadata=MetaData(name="user_service", team="fraud"),
    key_schema=StructType("keys", [StructField("user_id", LongType)]),
    value_schema=StructType("values", [
        StructField("country", StringType),
        StructField("phone_verified", BooleanType),
        StructField("account_age_days", IntType),
        StructField("total_purchases", IntType)
    ]),
    endpoint=RestEndpoint(
        url="https://user-api.company.com/v2/users/{user_id}",
        method=HttpMethod.GET,
        auth=AuthConfig(auth_type=AuthType.BEARER_TOKEN, token="${env.USER_TOKEN}"),
        response_mapping={
            "country": "$.user.location.country",
            "phone_verified": "$.user.verification.phone",
            "account_age_days": "$.user.account.age_days",
            "total_purchases": "$.user.stats.purchase_count"
        },
        timeout_ms=2000
    )
)
```

--------------------------------

### User Service GET Request

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Defines a REST external source for the user service. It uses a GET request with Bearer Token authentication to retrieve user details based on a user ID. Response mapping is provided to extract specific fields from the JSON response.

```APIDOC
## GET /v2/users/{user_id}

### Description
Retrieves user details including country, phone verification status, account age, and total purchases.

### Method
GET

### Endpoint
https://user-api.company.com/v2/users/{user_id}

### Parameters
#### Path Parameters
- **user_id** (string) - Required - The unique identifier for the user.

#### Request Body
This endpoint does not have a request body.

### Response
#### Success Response (200)
- **country** (string) - The country associated with the user.
- **phone_verified** (boolean) - Indicates if the user's phone number has been verified.
- **account_age_days** (integer) - The age of the user's account in days.
- **total_purchases** (integer) - The total number of purchases made by the user.

### Request Example
(No explicit request example provided in source)

### Response Example
(No explicit response example provided in source)
```

--------------------------------

### GraphQL External Source Configuration

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Sets up a GraphQL external source for fetching product catalog data. It includes a GraphQL query, variable mapping, and response mapping to Zipline schemas.

```python
product_catalog = GraphQLExternalSource(
    metadata=MetaData(name="product_catalog", team="retail"),

    key_schema=StructType("keys", [
        StructField("product_id", StringType)
    ]),

    value_schema=StructType("values", [
        StructField("name", StringType),
        StructField("category", StringType),
        StructField("price", DoubleType),
        StructField("in_stock", BooleanType),
        StructField("rating", DoubleType)
    ]),

    endpoint=GraphQLEndpoint(
        url="https://catalog-api.company.com/graphql",

        # GraphQL query with variables
        query="""
            query GetProduct($productId: ID!) {
                product(id: $productId) {
                    name
                    category {
                        name
                    }
                    pricing {
                        current
                    }
                    inventory {
                        available
                    }
                    reviews {
                        averageRating
                    }
                }
            }
        """,

        # Map key_schema fields to GraphQL variables
        variable_mapping={
            "productId": "product_id"
        },

        # Map GraphQL response to value_schema fields
        response_mapping={
            "name": "$.data.product.name",
            "category": "$.data.product.category.name",
            "price": "$.data.product.pricing.current",
            "in_stock": "$.data.product.inventory.available",
            "rating": "$.data.product.reviews.averageRating"
        },

        auth=AuthConfig(
            auth_type=AuthType.BEARER_TOKEN,
            token="${env.CATALOG_TOKEN}"
        ),

        timeout_ms=3000
    )
)
```

--------------------------------

### Run Join

Source: https://zipline.ai/docs/Kaggle_Outbrain

Executes a compiled join in backfill mode to generate a training dataset. Configure memory and parallelism settings appropriately. Sufficient RAM is crucial to prevent job failures.

```bash
DRIVER_MEMORY=4G EXECUTOR_MEMORY=8G EXECUTOR_CORES=6 PARALLELISM=100 MAX_EXECUTORS=1 
python3 ~/repos/chronon/api/python/ai/chronon/repo/run.py --mode=backfill 
--conf=production/joins/kaggle/outbrain.training_set 
--local-data-path ~/kaggle_outbrain --local-warehouse-location ~/kaggle_outbrain_parquet 
--ds=2016-07-01 --step-days=1
```

--------------------------------

### Request-Level Error Handling Example

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Shows how Chronon creates error features when an external API request fails, including exception details.

```python
user_service = RestExternalSource(
    metadata=MetaData(name="user_service"),
    # ... configuration ...
)
 
# If request fails, you get:
# - ext_user_service_country = null
# - ext_user_service_phone_verified = null
# - ext_user_service_exception = "Connection timeout after 3000ms"

```

--------------------------------

### Plan for Bootstrap with External Sources

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Configures a Join to use logged values for offline training by setting `bootstrap_from_log=True` and specifying `row_ids` for joining logged data with external features.

```python
fraud_features = Join(
    # ...
    online_external_parts=[ExternalPart(user_service)],
    row_ids=["event_id"],           # For joining logged data
    bootstrap_from_log=True          # Use logged values for training
)
```

--------------------------------

### Enable Batching for API Calls

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Demonstrates how to configure batching to reduce the number of API calls. By setting a max batch size, multiple individual requests can be consolidated into a single API call.

```python
# 100 individual requests = 100 API calls
# With batch_config(max_batch_size=100) = 1 API call
```

--------------------------------

### Run Staging Query

Source: https://zipline.ai/docs/Kaggle_Outbrain

Executes a compiled staging query in backfill mode to produce an output table. Adjust memory and parallelism settings based on your machine's resources. Ensure sufficient RAM to avoid OutOfMemory errors.

```bash
mkdir ~/kaggle_outbrain

DRIVER_MEMORY=2G EXECUTOR_CORES=6 EXECUTOR_MEMORY=8G PARALLELISM=10 MAX_EXECUTORS=1 
python3 ~/repos/chronon/api/python/ai/chronon/repo/run.py --mode=backfill 
--conf=production/staging_queries/kaggle/outbrain.base_table 
--local-data-path ~/kaggle_outbrain --local-warehouse-location ~/kaggle_outbrain_parquet
```

--------------------------------

### External Source Request-level Failure Handling

Source: https://zipline.ai/docs/authoring_features/ExternalSource

Example of an ExternalSourceHandler implementing fetch logic that recovers from TimeoutException by returning a Failure, and handles NotFoundException by returning null values.

```scala
class UserServiceHandler extends ExternalSourceHandler {
  override def fetch(requests: Seq[Request]): Future[Seq[Response]] = {
    Future.sequence(requests.map { request =>
      callUserService(request).recover {
        case ex: TimeoutException =>
          // Return failure - Chronon will create ext_user_service_exception
          Response(request, Failure(ex))
        case ex: NotFoundException =>
          // Return null values instead of error
          Response(request, Success(Map(
            "country" -> null,
            "phone_verified" -> null,
            "account_age_days" -> null
          )))
      }
    })
  }
}
```

--------------------------------

### Enable Online Serving for GroupBy

Source: https://zipline.ai/docs/authoring_features/GroupBy

Use the 'online=True' parameter to schedule pipelines for online low-latency serving. Once set, this cannot be updated without recompiling.

```python
your_gb = GroupBy(
  ...,
  online=True
)
```

--------------------------------

### Using StagingQuery in a Join

Source: https://zipline.ai/docs/authoring_features/StagingQuery

Demonstrates how to reference a previously defined StagingQuery within the 'left' side of a Chronon Join.

```python
from staging_queries.team_name_folder import file_name

v1 = Join(
    left=EventSource(
        table=file_name.staging_query_var_name.table
        ...
    )
)
```

--------------------------------

### Use Environment Variables for Secrets

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Illustrates the secure practice of using environment variables for API tokens instead of hardcoding them directly in the configuration.

```python
# Good
token="${env.API_TOKEN}"

# Bad
token="sk_live_abcd1234..."
```

--------------------------------

### Configuring a Time-Partitioned Source

Source: https://zipline.ai/docs/authoring_features/Source

Use this configuration when your data warehouse table has a timestamp or date column for partitioning instead of Hive-style string partitions. The `partition_column` is used for filtering, while `time_column` handles event-time logic.

```python
source = Source(
    events=EventSource(
        table="warehouse.user_events",
        query=Query(
            selects=select("user_id", "event_value"),
            time_column="UNIX_TIMESTAMP(event_ts) * 1000",  # millis for event-time logic
            partition_column="event_ts",  # the timestamp/date column to filter by
            time_partitioned=True,        # signals this is not a Hive-partitioned table
        )
    ))

```

--------------------------------

### Configure Step Days for StagingQuery

Source: https://zipline.ai/docs/authoring_features/StagingQuery

Use the `customJson` argument with `additional_args` to set a maximum number of days for a StagingQuery run. This helps manage job size for long backfills.

```python
v1 = StagingQuery(
    ...
    customJson=json.dumps({
            "additional_args": ["--step-days=30"], # Sets the maximum days to run in one job to 30
        }),
)
```

--------------------------------

### Adjust JSONPath Mapping for Response

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Provides an example of correcting a JSONPath mapping to accurately extract data from an API response. It shows the difference between a wrong and a correct path for retrieving country information.

```python
# Wrong: "$.user.country"
# Correct: "$.data.user.location.country"
response_mapping={
    "country": "$.data.user.location.country"
}
```

--------------------------------

### Utilize Online External Features

Source: https://zipline.ai/docs/authoring_features/Join

Configures a join to include external features fetched online, such as contextual information or features from external services. It also shows how to bootstrap historical data for these external fields.

```python
v1 = Join(
   left=...,
   right_parts=...,
   online_external_parts=[
      # contextual fields including request id and request context string
      ExternalPart(ContextualSource(
         fields=[
            ("event_id", DataType.STRING),
            ("request_context", DataType.STRING)
         ]
      )),
      # external feature definition that calls userService to fetch user features
      ExternalPart(ExternalSource(
         name="service_user",
         team="team_name",
         key_fields=[
            ("user", DataType.LONG)
         ],
         value_fields=[
            ("country", DataType.STRING),
            ("phone_country", DataType.STRING),
      )))
   ],
   # bootstrap_part to ingest historical data for external fields
   bootstrap_parts=[
      BootstrapPart(
         table="<custom_external_backfill_table>",
         query=Query(
            # selected field names need to match feature names
            selects=select(
               request_context_id="...",
               ext_service_user_country="...",
               ext_service_user_phone_country="...",
            ),
            start_partition="...",
            end_partition="..."
         )
      )
   ]
)
```

--------------------------------

### Compute Consistency Metrics

Source: https://zipline.ai/docs/getting_started/Local_Sandbox

Calculate online/offline consistency metrics by comparing backfilled feature values against actual logged online fetches. This command requires the join configuration.

```bash
run.py --mode consistency-metrics-compute --conf production/joins/quickstart/training_set.v2
```

--------------------------------

### StagingQuery with Time-Partitioned Input Table

Source: https://zipline.ai/docs/authoring_features/StagingQuery

Configure a StagingQuery to read from a time-partitioned table by setting `time_partitioned=True` in the `TableDependency`. The query must produce a `ds` column. Use `{{ end_date(offset=1) }}` for exclusive upper bounds in timestamp comparisons.

```python
v1 = StagingQuery(
    query="""
        SELECT
            user_id,
            event_value,
            DATE(event_ts) as ds
        FROM warehouse.user_events
        WHERE event_ts >= {{ start_date }}
          AND event_ts < {{ end_date(offset=1) }}
    """,
    metaData=MetaData(
        dependencies=[
            TableDependency(
                table_info=TableInfo(
                    table="warehouse.user_events",
                    partition_column="event_ts",
                    time_partitioned=True,
                ),
            ),
        ],
    ),
)
```

--------------------------------

### Configure Retries for Requests

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Configures retry mechanisms for API requests. It recommends setting max_retries for idempotent GET requests and disabling retries for non-idempotent POST/PUT requests to avoid unintended side effects.

```python
# For idempotent GET requests
retry=RetryConfig(max_retries=3)

# For non-idempotent POST/PUT (use with caution)
retry=RetryConfig(max_retries=0)
```

--------------------------------

### Migrating Hardcoded Secrets: Before

Source: https://zipline.ai/docs/authoring_features/SecretsManagement

Illustrates the structure of configuration before migrating to vault URIs. This includes environment variables and Spark configuration with hardcoded credentials.

```json
env={
    "SNOWFLAKE_VAULT_URI": "https://demo-service-writer-pkey.vault.azure.net/secrets/snowflake-private-key",
},
conf={
    "spark.sql.catalog.spark_catalog.credential": "XtyCirtE0/o3pcTMdkLCh7LXVno=:i++cOG/+..."
}
```

--------------------------------

### SCD2 GroupBy Configuration

Source: https://zipline.ai/docs/authoring_features/SCD2

Configure an SCD2 GroupBy to aggregate historical data based on validity intervals. This example sets up sources, keys, and various aggregation operations like SUM, AVG, COUNT, and COUNT_DISTINCT.

```python
from ai.chronon import scd2
from ai.chronon.types import EventSource, Query

contracts_source = EventSource(
    table="warehouse.contracts_scd2",
    query=Query(
        selects={
            "user_id": "user_id",
            "contract_id": "contract_id",
            "valid_from_ts": "valid_from_ts",
            "valid_to_ts": "valid_to_ts",
            "amount": "amount",
            "status": "status",
        },
        time_column="valid_from_ts",
    ),
)

contracts = scd2.GroupBy(
    sources=[contracts_source],
    keys=["user_id"],
    row_id_column="contract_id",
    valid_from_column="valid_from_ts",
    valid_to_column="valid_to_ts",
    aggregations=[
        scd2.Aggregation(
            input_column="amount",
            operation=scd2.Operation.SUM,
            windows=["7d", "30d"],
        ),
        scd2.Aggregation(
            input_column="amount",
            operation=scd2.Operation.AVG,
        ),
        scd2.Aggregation(
            operation=scd2.Operation.COUNT,
        ),
        scd2.Aggregation(
            operation=scd2.Operation.EXISTS,
        ),
        scd2.Aggregation(
            input_column="status",
            operation=scd2.Operation.COUNT_DISTINCT,
        ),
        scd2.Aggregation(
            operation=scd2.Operation.COUNT,
            windows=["10950d"],
            window_column="birth_ts",
        ),
    ],
)
```

--------------------------------

### Log-based Bootstrap Strategy

Source: https://zipline.ai/docs/authoring_features/ExternalSource

Configures a Join operation to automatically use logged online request data for bootstrapping historical external features.

```python
v1 = Join(
    left=...,
    right_parts=[...],
    online_external_parts=[
        ExternalPart(ContextualSource(fields=[("event_id", StringType)])),
        ExternalPart(user_service_source)
    ],
    row_ids=["event_id"],
    # Automatically use logged values for bootstrap
    bootstrap_from_log=True
)
```

--------------------------------

### Basic Authentication

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Configure a RestEndpoint for Basic Authentication using username and password from environment variables.

```python
endpoint=RestEndpoint(
    url="https://api.company.com/resource/{id}",
    auth=AuthConfig(
        auth_type=AuthType.BASIC_AUTH,
        username="${env.API_USERNAME}",
        password="${env.API_PASSWORD}"
    )
)
```

```http
GET /resource/123 HTTP/1.1
Host: api.company.com
Authorization: Basic dXNlcm5hbWU6cGFzc3dvcmQ=
```

--------------------------------

### Before: Custom Handler Configuration

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Shows the configuration of an `ExternalSource` using a custom handler, including metadata, key schema, and value schema, alongside a separate Scala handler implementation.

```scala
# Configuration
user_service = ExternalSource(
    metadata=MetaData(name="user_service"),
    key_schema=StructType("keys", [StructField("user_id", LongType)])
    value_schema=StructType("values", [...])
)

# Separate Scala handler implementation (50+ lines)
class UserServiceHandler extends ExternalSourceHandler {
  // Custom HTTP code...
}
```

--------------------------------

### Compile and Backfill Features

Source: https://zipline.ai/docs/getting_started/Local_Sandbox

Compile feature schemas and backfill features using Chronon's run.py script. This prepares the data for consistency checks.

```bash
compile.py --conf group_bys/quickstart/schema.py
```

```bash
run.py --mode backfill --conf production/group_bys/quickstart/schema.v1
```

--------------------------------

### Hybrid Bootstrap Strategy

Source: https://zipline.ai/docs/authoring_features/ExternalSource

Combines log-based bootstrapping for recent data with a custom backfill table for older historical data.

```python
v1 = Join(
    left=driver_table_with_union_history,  # Union of old and new drivers
    online_external_parts=[...],
    # Bootstrap from log for recent data
    bootstrap_from_log=True,
    # Also include custom backfill for older data
    bootstrap_parts=[
        BootstrapPart(
            table="my_namespace.historical_external_features",
            query=Query(
                # Only use for dates before logging started
                wheres=["ds < '2023-06-01'"],
                selects={...}
            )
        )
    ]
)
```

--------------------------------

### Set up Log-Based Data Refresh for Online Model

Source: https://zipline.ai/docs/authoring_features/Join

Configures a feature pipeline to automatically populate new data using logged feature values. This is crucial for online models to ensure accurate and up-to-date feature serving.

```python
# ml_models/zipline/staging_queries/team_name/driver_table.py
v1 = StagingQuery(
  query="""
  WITH
  log_drivers AS (
    SELECT event_id, <entity_keys>, ts, ds
    FROM (
      SELECT
       event_id, <entity_keys>, ts, ds
       , ROW_NUMBER() OVER (PARTITION BY event_id, ds ORDER BY ts DESC) AS rank
      FROM <db_name>.<join_name>_logged
      WHERE ds BETWEEN '{{ start_date }}' AND '{{ end_date }}'
      AND ds > '<historical_data_end_date>'
    ) a
    WHERE rank = 1
  ),
  init_drivers AS (
    SELECT event_id, <entity_keys>, ts, ds
    FROM <db_name>.<driver_table_before_logging>
    WHERE ds BETWEEN '{{ start_date }}' AND '{{ end_date }}'
    AND ds <= '<historical_data_end_date>'
  )
  SELECT *
  FROM log_drivers
  UNION ALL
  SELECT *
  FROM init_drivers
  """)
```

--------------------------------

### Configure BatchConfig with Result Key Field

Source: https://zipline.ai/docs/authoring_features/ExternalSource_REST

Illustrates setting the `result_key_field` in `BatchConfig` to correctly map batch response items to their corresponding request keys, resolving issues where batch responses don't match requests.

```python
# Request key: "item_id"
# Response: [{"id": 101, ...}, {"id": 102, ...}]

batch_config=BatchConfig(
    result_key_field="id"  # Must match "id" in response
)
```

--------------------------------

### Define Production Event Source (v1)

Source: https://zipline.ai/docs/authoring_features/Bootstrap

Creates an EventSource for the v1 join, selecting key entity identifiers and a random number generator for potential downsampling. This source is based on the 'payments_driver.v1' table.

```python
ENTITY_KEYS = ['guest', 'host', 'listing']

v1_source = EventSource(
	table=payments_driver.v1.table,
	query=Query(
		selects=select(
			rng='rng',
			**{pk: pk for pk in ENTITY_KEYS}
		)
	)
)
```
Tokens:
Raw