Node Types

Reference for all 15 node types. Each section covers config options, input/output, and examples.

Overview

source_file

Read data from CSV, JSON, Parquet, or Excel files.

Config:

Example:

{
  "path": "/data/sales.csv",
  "format": "csv",
  "delimiter": ","
}

File security: Set BROKOLI_DATA_DIRS to restrict which directories Brokoli can read. Default: unrestricted.

source_api

Fetch data from a REST API endpoint.

Config:

Example:

{
  "url": "https://api.example.com/users",
  "method": "GET",
  "headers": {"Authorization": "Bearer ${var.api_token}"}
}

source_db

Run a SQL query against a database connection.

Config:

Example:

{
  "conn_id": "prod-postgres",
  "query": "SELECT id, name, email FROM users WHERE active = true"
}

transform

Apply one or more transformation rules to the incoming dataset. Rules execute in order.

Config:

Transform rule types

Short aliases: You can use short aliases instead of the full rule names: rename for rename_columns, filter for filter_rows, function for apply_function, replace for replace_values, drop for drop_columns, dedup for deduplicate, agg for aggregate.

Aggregate example:

{
  "rules": [{
    "type": "aggregate",
    "group_by": ["department"],
    "agg_fields": [
      {"column": "salary", "function": "avg", "alias": "avg_salary"},
      {"column": "id", "function": "count", "alias": "headcount"}
    ]
  }]
}

Supported aggregate functions: sum, count, avg, min, max.

Filter example:

{
  "rules": [
    {"type": "filter_rows", "condition": "status = active"},
    {"type": "filter_rows", "condition": "age != 0"},
    {"type": "filter_rows", "condition": "country in [US, UK, DE]"}
  ]
}

code

Run a custom Python script. The script receives columns (list), rows (list of dicts), config (dict), and params (dict). Set output_data to return results.

Config:

Example:

{
  "script": "rows = [r for r in rows if r.get('score', 0) > 50]\noutput_data = {'columns': columns, 'rows': rows}",
  "timeout": 60
}

For datasets over 10,000 rows, Brokoli automatically switches to file-based transfer (NDJSON or CSV) for faster processing. If pyarrow or pandas are installed, they're used automatically.

join

Join two input datasets on a common key.

Config:

Connect two upstream nodes to a join node. The first input is the left table, the second is the right table.

Example:

{
  "left_key": "user_id",
  "right_key": "user_id",
  "join_type": "left"
}

sql_generate

Generate SQL statements from the input dataset. Useful for creating bulk INSERT or UPSERT scripts.

Config:

quality_check

Assert data quality rules. See Quality Checks for full details.

Config:

Example:

{
  "checks": [
    {"column": "email", "rule": "not_null", "on_failure": "block"},
    {"column": "id", "rule": "unique", "on_failure": "warn"}
  ]
}

condition

Branch pipeline execution based on data. See Conditions for details.

Config:

sink_file

Write the input dataset to a file.

Config:

sink_db

Write rows to a database table.

Config:

Example:

{
  "conn_id": "warehouse-pg",
  "table": "dim_users",
  "mode": "upsert",
  "conflict_key": "user_id"
}

sink_api

POST the dataset to a REST API endpoint.

Config:

dbt

Run dbt CLI commands and capture structured results. Requires dbt-core installed and accessible on PATH.

Config:

Output: A dataset with columns model, status, execution_time, message parsed from dbt's JSON output.

Example:

{
  "command": "run",
  "project_dir": "/opt/dbt/my_project",
  "target": "prod",
  "select": "tag:daily"
}

Runs dbt <command> --project-dir <project_dir> --output json. If JSON parsing fails, falls back to reading target/run_results.json.

notify

Send a notification via Slack or a generic webhook. This is a pass-through node -- it returns the input data unchanged.

Config:

Message templates: Use {{pipeline}}, {{run_id}}, and {{rows}} as placeholders in your message string. They are replaced at runtime with the pipeline name, current run ID, and input row count.

Slack mode (notify_type: "slack"): Sends {"text": "<message>", "channel": "<channel>"} to the webhook URL.

Webhook mode (notify_type: "webhook"): Sends {"pipeline": "<name>", "run_id": "<id>", "message": "<message>", "row_count": <n>, "timestamp": "<iso8601>"} to the webhook URL.

Example:

{
  "notify_type": "slack",
  "webhook_url": "https://hooks.slack.com/services/T00/B00/xxx",
  "message": "Pipeline {{pipeline}} finished with {{rows}} rows",
  "channel": "#data-alerts"
}

migrate

Run database schema migrations.

Config:

Example:

{
  "conn_id": "prod-postgres",
  "sql": "ALTER TABLE users ADD COLUMN last_login TIMESTAMP;"
}

Common config keys

These keys work on any node type: