Bruin

https://getbruin.com/docs
Admin
Bruin is an end-to-end data framework for ingestion, transformations (SQL, Python, R), and quality...
Tokens:25,676
Snippets:358
Trust Score:6.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install DuckDB

Source: https://getbruin.com/docs/ingestr/tutorials/load-personio-duckdb.html

Install the DuckDB library using pip.

```bash
pip install duckdb
```

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

### Install ingestr via script

Source: https://getbruin.com/docs/ingestr

Use the provided shell script to install ingestr on your system.

```sh
curl -LsSf https://getbruin.com/install/ingestr | sh
```

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

### Start the ingestr server

Source: https://getbruin.com/docs/ingestr/commands/server.html

Launches the web UI on the default port 8080.

```bash
ingestr server
```

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

### Example table output

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Representation of the table structure after an initial run.

```text
product_id | product_name | stock | _is_current | _valid_from         | _valid_until
1          | Laptop       | 100   | true        | 2025-04-02 00:00:00| 9999-12-31 23:59:59
2          | Smartphone   | 150   | true        | 2025-04-02 00:00:00| 9999-12-31 23:59:59
3          | Headphones   | 175   | true        | 2025-04-02 00:00:00| 9999-12-31 23:59:59
4          | Monitor      | 25    | true        | 2025-04-02 00:00:00| 9999-12-31 23:59:59
```

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

### Example CSV Data

Source: https://getbruin.com/docs/bruin/platforms/motherduck.html

Sample CSV content for a seed asset.

```csv
name,networking_through,position,contact_date
Y,LinkedIn,SDE,2024-01-01
B,LinkedIn,SDE 2,2024-01-01
```

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

### POST /api/run

Source: https://getbruin.com/docs/ingestr/commands/server.html

Starts an ingestion run.

```APIDOC
## POST /api/run

### Description
Starts an ingestion run.

### Method
POST

### Endpoint
/api/run
```

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

### Configure server with CLI flags

Source: https://getbruin.com/docs/ingestr/commands/server.html

Starts the server using custom paths and ports via command-line arguments.

```bash
ingestr server \
  --port 3000 \
  --creds-file ~/.ingestr/creds.json \
  --logs-dir ~/.ingestr/logs \
  --db ~/.ingestr/ingestr.db
```

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

### Install ingestr

Source: https://getbruin.com/docs/ingestr/tutorials/load-stripe-postgres.html

Command to install the ingestr package via pip.

```bash
pip install ingestr
```

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

### Define columns and quality checks

Source: https://getbruin.com/docs/bruin/assets/columns.html

Example showing how to define columns with types, descriptions, primary keys, and various quality checks.

```yaml
columns:
  - name: col1
    type: integer
    description: "Just a number"
    primary_key: true
    checks:
      - name: unique
      - name: not_null
      - name: positive
  - name: col2
    type: string
    description: |
      some multi-line definition for this column
    update_on_merge: true
    checks:
      - name: not_null
      - name: accepted_values
        value: [ 'value1', 'value2' ]
```

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

### Define an asset with scd2_by_time materialization

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Example configuration for a BigQuery SQL asset using the scd2_by_time strategy.

```bruin-sql
/* @bruin
name: test.products
type: bq.sql

materialization:
  type: table
  strategy: scd2_by_time
  incremental_key: dt

columns:
  - name: product_id
    type: INTEGER
    description: "Unique identifier for the product"
    primary_key: true
  - name: product_name
    type: VARCHAR
    description: "Name of the product"
  - name: stock
    type: INTEGER
    description: "Number of units in stock"
  - name: dt
    type: DATE
    description: "Date when the product was last updated"
@bruin */

SELECT
    1 AS product_id,
    'Laptop' AS product_name,
    100 AS stock,
    DATE '2025-04-02' AS dt
UNION ALL
SELECT
    2 AS product_id,
    'Smartphone' AS product_name,
    150 AS stock,
    DATE '2025-04-02' AS dt
```

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

### Define supported dashboard tool assets

Source: https://getbruin.com/docs/bruin/assets/dashboard.html

Minimal configuration examples for various supported dashboard tools.

```yaml
name: myschema.asset_name
type: quicksight
```

```yaml
name: myschema.asset_name
type: superset
```

```yaml
name: myschema.asset_name
type: domo
```

```yaml
name: myschema.asset_name
type: gooddata
```

```yaml
name: myschema.asset_name
type: grafana
```

```yaml
name: myschema.asset_name
type: looker
```

```yaml
name: myschema.asset_name
type: looker_studio
```

```yaml
name: myschema.asset_name
type: metabase
```

```yaml
name: myschema.asset_name
type: modebi
```

```yaml
name: myschema.asset_name
type: powerbi
```

```yaml
name: myschema.asset_name
type: qliksense
```

```yaml
name: myschema.asset_name
type: qlikview
```

```yaml
name: myschema.asset_name
type: redash
```

```yaml
name: myschema.asset_name
type: sisense
```

```yaml
name: myschema.asset_name
type: tableau
```

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

### Execute Kinesis to BigQuery ingestion

Source: https://getbruin.com/docs/ingestr/tutorials/load-kinesis-bigquery.html

A complete example command to run the ingestion process from Kinesis to BigQuery.

```bash
ingestr ingest \   
    --source-uri 'kinesis://?aws_access_key_id=<YOUR_KEY_ID>&aws_secret_access_key=<YOUR_SECRET_KEY>&region_name=eu-central-1' \   
    --source-table 'kinesis_stream_name' \   
    --dest-uri 'bigquery://project-name?credentials_path=/Users/abc.json' \   
    --dest-table 'dataset.results'
```

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

### Define foreign key relationships

Source: https://getbruin.com/docs/bruin/assets/columns.html

Example showing how to document a foreign key relationship between assets and include a relationship quality check.

```yaml
columns:
  - name: customer_id
    type: integer
    foreign_key:
      table: customers   # the name of another asset
      column: id         # the referenced column in that asset
    checks:
      - name: relationships # optionally validate the relationship after the asset runs
```

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

### Define asset materialization

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Example of a BigQuery SQL asset configured with table materialization.

```sql
/* @bruin

name: dashboard.hello_bq
type: bq.sql

materialization:
    type: table

@bruin */

select 1 as one
union all
select 2 as one
```

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

### Monitor a stream with Prometheus metrics

Source: https://getbruin.com/docs/ingestr/commands/ingest.html

Starts an HTTP server to expose Prometheus metrics for a stream, allowing for real-time monitoring of replication lag and sync progress.

```bash
ingestr ingest \
   --source-uri 'postgres+cdc://user:pass@localhost:5432/mydb' \
   --dest-uri 'bigquery://my_project?credentials_path=/path/to/sa.json' \
   --stream \
   --metrics-addr 127.0.0.1:6060

curl -s localhost:6060/metrics | grep -E '^ingestr_replication|^ingestr_stream_table'
```

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

### Define SCD2 Asset

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Example of an asset definition using the scd2_by_column materialization strategy.

```sql
/* @bruin
name: test.product_catalog
type: bq.sql

materialization:
  type: table
  strategy: scd2_by_column
```

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

### Define a dashboard asset schema

Source: https://getbruin.com/docs/bruin/assets/dashboard.html

Example of a complete dashboard asset configuration including dependencies, owner, and tags.

```yaml
name: dashboard.quicksight
type: quicksight
description: "Dashboard sales data visualization."

depends:
    - schema.my_asset

owner: my-team@acme-corp.com

tags:
  - dashboard
  - team.xyz
```

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

### Sensor Example: Partitioned Upstream Table

Source: https://getbruin.com/docs/bruin/platforms/motherduck.html

Check for data availability in an upstream table based on the end date.

```yaml
name: analytics_123456789.events
type: duckdb.sensor.query
connection: my_motherduck_connection
parameters:
    query: select exists(select 1 from upstream_table where dt = "{{ end_date }}")
```

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

### Define Asset Hooks

Source: https://getbruin.com/docs/bruin/assets/definition-schema.html

Executes SQL snippets before or after the main asset query for setup or cleanup tasks.

```yaml
hooks:
  pre:
    - query: "INSTALL httpfs"
    - query: "LOAD httpfs"
  post:
    - query: "SET s3_region=''"
```

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

### GET /api/credentials

Source: https://getbruin.com/docs/ingestr/commands/server.html

Lists saved connections.

```APIDOC
## GET /api/credentials

### Description
Lists saved connections.

### Method
GET

### Endpoint
/api/credentials
```

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

### SQL Asset with Inline Definition

Source: https://getbruin.com/docs/bruin/assets/definition-schema.html

Example of a BigQuery SQL asset containing its metadata definition within a /* @bruin ... @bruin */ block.

```sql
/* @bruin

name: dashboard.hello_bq
type: bq.sql

owner: my-team@acme-corp.com

depends:
   - hello_python

materialization:
   type: table

rerun_cooldown: 300

tags:
   - dashboard
   - team:xyz

columns:
  - name: one
    type: integer
    description: "Just a number"
    checks:
        - name: unique
        - name: not_null
        - name: positive
        - name: accepted_values
          value: [1, 2]

@bruin */

select 1 as one
union all
select 2 as one
```

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

### Run continuous streaming ingestion

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-postgres-duckdb.html

Starts a long-running process to tail the Postgres WAL and flush changes to DuckDB at a specified interval.

```bash
ingestr ingest \
  --source-uri "postgres+cdc://user:password@localhost:5432/shop?sslmode=disable" \
  --source-table "public.customers" \
  --dest-uri "duckdb:///shop.duckdb" \
  --dest-table "public.customers" \
  --stream --flush-interval 2s
```

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

### Define Data Vault Satellite Query

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Example SQL query for a Data Vault satellite asset.

```sql
SELECT customer_hk, hashdiff, load_dts, record_source, customer_name, email
FROM stg.customer_hashed
```

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

### Sensor Example: Streaming Upstream Table

Source: https://getbruin.com/docs/bruin/platforms/motherduck.html

Check for new data in a streaming table based on the end timestamp.

```yaml
name: analytics_123456789.events
type: duckdb.sensor.query
connection: my_motherduck_connection
parameters:
    query: select exists(select 1 from upstream_table where inserted_at > "{{ end_timestamp }}")
```

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

### Standalone YAML Asset Definition

Source: https://getbruin.com/docs/bruin/assets/definition-schema.html

Example of a standalone YAML asset definition file, typically used for asset types without an inline code body.

```yaml
name: raw.hello_external
type: ingestr
owner: my-team@acme-corp.com

parameters:
  source_connection: my-source
  source_table: public.hello
  destination: bigquery
```

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

### Create Sample MySQL Table

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-mysql-duckdb.html

Set up a sample table with a primary key to demonstrate the replication process.

```sql
CREATE TABLE customers (
    id    INT PRIMARY KEY,
    name  VARCHAR(100) NOT NULL,
    email VARCHAR(255)
);
INSERT INTO customers (id, name, email) VALUES
    (1, 'Alice', 'alice@example.com'),
    (2, 'Bob',   'bob@example.com'),
    (3, 'Carol', 'carol@example.com');
```

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

### Create and populate source table

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-postgres-duckdb.html

Set up a customers table with a primary key to serve as the source for replication.

```sql
CREATE TABLE public.customers (
    id    INTEGER PRIMARY KEY,
    name  TEXT NOT NULL,
    email TEXT
);
INSERT INTO public.customers (id, name, email) VALUES
    (1, 'Alice', 'alice@example.com'),
    (2, 'Bob',   'bob@example.com'),
    (3, 'Carol', 'carol@example.com');
```

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

### Server startup output

Source: https://getbruin.com/docs/ingestr/commands/server.html

The console output displayed when the server initializes successfully.

```text
Starting ingestr web UI on http://localhost:8080
```

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

### GET /api/runs/{id}

Source: https://getbruin.com/docs/ingestr/commands/server.html

Returns a persisted run.

```APIDOC
## GET /api/runs/{id}

### Description
Returns a persisted run.

### Method
GET

### Endpoint
/api/runs/{id}

### Parameters
#### Path Parameters
- **id** (string) - Required - The ID of the run.
```

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

### Perform initial full refresh

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Command to initialize the SCD2 table if it does not exist.

```bash
bruin run --full-refresh path/to/your/products.sql
```

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

### GET /api/runs

Source: https://getbruin.com/docs/ingestr/commands/server.html

Lists persisted run history.

```APIDOC
## GET /api/runs

### Description
Lists persisted run history. Supports limit and offset.

### Method
GET

### Endpoint
/api/runs

### Parameters
#### Query Parameters
- **limit** (integer) - Optional - Number of records to return.
- **offset** (integer) - Optional - Number of records to skip.
```

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

### GET /api/runs/{id}/logs

Source: https://getbruin.com/docs/ingestr/commands/server.html

Returns persisted logs for a run.

```APIDOC
## GET /api/runs/{id}/logs

### Description
Returns persisted logs for a run.

### Method
GET

### Endpoint
/api/runs/{id}/logs

### Parameters
#### Path Parameters
- **id** (string) - Required - The ID of the run.
```

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

### GET /api/connectors

Source: https://getbruin.com/docs/ingestr/commands/server.html

Lists connector metadata used by the UI forms.

```APIDOC
## GET /api/connectors

### Description
Lists connector metadata used by the UI forms.

### Method
GET

### Endpoint
/api/connectors
```

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

### Run Initial Ingestr Load

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-mysql-duckdb.html

Execute the initial snapshot from MySQL to DuckDB using the mysql+cdc scheme.

```bash
ingestr ingest \
  --source-uri "mysql+cdc://ingestr_cdc:cdcpass@localhost:3306/shop?server_id=1001" \
  --source-table "customers" \
  --dest-uri "duckdb:///warehouse.duckdb" \
  --dest-table "shop.customers"
```

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

### GET /api/ws/logs/{id}

Source: https://getbruin.com/docs/ingestr/commands/server.html

Streams live logs for a running job over WebSocket.

```APIDOC
## GET /api/ws/logs/{id}

### Description
Streams live logs for a running job over WebSocket.

### Method
GET

### Endpoint
/api/ws/logs/{id}

### Parameters
#### Path Parameters
- **id** (string) - Required - The ID of the job.
```

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

### Import QuickSight Assets via CLI

Source: https://getbruin.com/docs/bruin/assets/quicksight-refresh.html

Use the Bruin import command to discover and generate asset definitions from your AWS QuickSight account.

```bash
bruin import quicksight ./my-pipeline --connection quicksight-prod
```

```bash
bruin import quicksight ./my-pipeline --connection quicksight-prod --all
```

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

### GET /api/jobs/{id}

Source: https://getbruin.com/docs/ingestr/commands/server.html

Returns the in-memory status for an active or recently completed job.

```APIDOC
## GET /api/jobs/{id}

### Description
Returns the in-memory status for an active or recently completed job.

### Method
GET

### Endpoint
/api/jobs/{id}

### Parameters
#### Path Parameters
- **id** (string) - Required - The ID of the job.
```

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

### Configure server with environment variables

Source: https://getbruin.com/docs/ingestr/commands/server.html

Sets server configuration using environment variables before executing the command.

```bash
export INGESTR_PORT=3000
export INGESTR_CREDS_FILE=~/.ingestr/creds.json
export INGESTR_LOGS_DIR=~/.ingestr/logs
export INGESTR_DB=~/.ingestr/ingestr.db

ingestr server
```

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

### Perform Initial Ingestion

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-sqlserver-duckdb.html

Execute the ingestr command to snapshot the SQL Server table into a DuckDB file.

```bash
ingestr ingest \
  --source-uri "mssql+cdc://user:password@localhost:1433/shop?encrypt=disable" \
  --source-table "dbo.customers" \
  --dest-uri "duckdb:///warehouse.duckdb" \
  --dest-table "dbo.customers"
```

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

### Configure QuickSight Connection

Source: https://getbruin.com/docs/bruin/assets/quicksight-refresh.html

Define the QuickSight connection in the .bruin.yml file using standard AWS credentials or temporary session tokens.

```yaml
connections:
  quicksight:
    - name: "connection_name"
      aws_access_key_id: "your-access-key-id"
      aws_secret_access_key: "your-secret-access-key"
      aws_region: "us-east-1"
      aws_account_id: "123456789012"
```

```yaml
connections:
  quicksight:
    - name: "connection_name"
      aws_access_key_id: "your-access-key-id"
      aws_secret_access_key: "your-secret-access-key"
      aws_session_token: "your-session-token"
      aws_region: "us-east-1"
      aws_account_id: "123456789012"
```

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

### Execute Personio to DuckDB ingestion

Source: https://getbruin.com/docs/ingestr/tutorials/load-personio-duckdb.html

Run the ingestion process using Personio credentials and a DuckDB destination file.

```bash
ingestr ingest \
    --source-uri 'personio://?client_id=<YOUR_CLIENT_ID>&client_secret=<YOUR_CLIENT_SECRET>' \
    --source-table 'employees' \
    --dest-uri 'duckdb:///personio.duckdb' \
    --dest-table 'dest.employees'
```

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

### Define static interval modifiers

Source: https://getbruin.com/docs/bruin/assets/interval-modifiers.html

Use the interval_modifiers block to shift the start and end times of the processing window.

```yaml
interval_modifiers:
  start: -2h    # Shift start time back 2 hours
  end: 1h       # Shift end time forward 1 hour
```

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

### Enable CDC on SQL Server

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-sqlserver-duckdb.html

Configure the database and table for log-based CDC and insert initial sample data.

```sql
-- enable CDC for the database
USE shop;
EXEC sys.sp_cdc_enable_db;

-- our demo table (any table with a primary key works)
CREATE TABLE dbo.customers (
    id    INT PRIMARY KEY,
    name  NVARCHAR(100) NOT NULL,
    email NVARCHAR(255)
);

-- enable CDC for the table; this creates the capture instance "dbo_customers"
EXEC sys.sp_cdc_enable_table
    @source_schema = 'dbo',
    @source_name   = 'customers',
    @role_name     = NULL;

INSERT INTO dbo.customers (id, name, email) VALUES
    (1, 'Alice', 'alice@example.com'),
    (2, 'Bob',   'bob@example.com'),
    (3, 'Carol', 'carol@example.com');
```

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

### Create View with MotherDuck

Source: https://getbruin.com/docs/bruin/platforms/motherduck.html

Define a SQL asset to create a view in MotherDuck.

```sql
/* @bruin
name: orders_per_country
type: duckdb.sql
connection: my_motherduck_connection
materialization:
    type: view
@bruin */

SELECT COUNT(*) as orders, country
FROM events.orders
WHERE status = "paid"
GROUP BY country
```

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

### Run a streaming asset

Source: https://getbruin.com/docs/bruin/assets/ingestr.html

Execute a single streaming asset in the foreground using the --stream flag.

```bash
bruin run --stream assets/my_stream.asset.yml
```

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

### Configure Asset-Level Interval Modifiers

Source: https://getbruin.com/docs/bruin/assets/interval-modifiers.html

Apply specific start and end time shifts directly within an asset definition.

```yaml
name: example.products
type: bq.sql

interval_modifiers:
  start: -2h    # Shift start time back 2 hours
  end: 1h       # Shift end time forward 1 hour

columns:
  - name: product_id
    type: INTEGER
    primary_key: true
  - name: created_at
    type: TIMESTAMP
```

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

### Configure QuickSight Dataset Refresh

Source: https://getbruin.com/docs/bruin/assets/quicksight-refresh.html

Basic configuration to trigger a SPICE refresh for a QuickSight dataset.

```yaml
name: quicksight.datasets.dataset_issues_custom_sql
type: quicksight.dataset
description: 'QuickSight dataset: issues_custom_sql'

parameters:
  dataset_id: 23e4f645-9837-4e73-ad15-04ccd4baa400
  dataset_name: issues_custom_sql
  import_mode: SPICE
  refresh: "true"
  custom_sql: "select * from issues where true limit 50"

columns:
  - name: id
    type: STRING
  - name: title
    type: STRING
  - name: description
    type: STRING
```

```yaml
name: quicksight.datasets.dataset_issues_custom_sql
type: quicksight.dataset
description: 'QuickSight dataset: issues_custom_sql'

parameters:
  dataset_id: 23e4f645-9837-4e73-ad15-04ccd4baa400
  dataset_name: issues_custom_sql
  import_mode: SPICE
  refresh: "true"
  incremental: "true"
  refresh_timeout_minutes: "90"
  custom_sql: "select * from issues where true limit 50"

columns:
  - name: id
    type: STRING
  - name: title
    type: STRING
  - name: description
    type: STRING
```

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

### Define Pipeline-Level Interval Modifiers

Source: https://getbruin.com/docs/bruin/assets/interval-modifiers.html

Set default start and end shifts for all assets within a pipeline using the default section.

```yaml
default:
  interval_modifiers:
    start: 2h    # Default 2-hour shift for all assets
    end: 2h      # Default 2-hour shift for all assets
```

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

### Run initial CDC snapshot

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-postgres-duckdb.html

Executes an initial snapshot of a Postgres table into DuckDB using the postgres+cdc scheme.

```bash
ingestr ingest \
  --source-uri "postgres+cdc://user:password@localhost:5432/shop?sslmode=disable" \
  --source-table "public.customers" \
  --dest-uri "duckdb:///shop.duckdb" \
  --dest-table "public.customers"
```

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

### Check MySQL Binary Log Settings

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-mysql-duckdb.html

Verify that binary logging is enabled and configured correctly for CDC.

```sql
SELECT @@log_bin, @@binlog_format, @@binlog_row_image;
```

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

### Configure asset parameters

Source: https://getbruin.com/docs/bruin/assets/definition-schema.html

Set asset-specific configurations such as full_refresh to control materialization behavior.

```yaml
parameters:
  full_refresh: true
```

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

### Copy MySQL to BigQuery

Source: https://getbruin.com/docs/bruin/assets/ingestr.html

Basic configuration for ingesting a table from a MySQL source to a BigQuery destination.

```yaml
name: raw.transactions
type: ingestr
parameters:
  source_connection: mysql_prod
  source_table: public.transactions
  destination: bigquery
```

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

### Configure materialization schema

Source: https://getbruin.com/docs/bruin/assets/materialization.html

YAML configuration for defining table materialization strategies, partitioning, and clustering.

```yaml
materialization:
  type: table
  strategy: delete+insert
  incremental_key: dt
  partition_by: dt
  cluster_by:
    - dt
    - user_id
```

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

### Copy MSSQL to Snowflake incrementally

Source: https://getbruin.com/docs/bruin/assets/ingestr.html

Configure an incremental load from Microsoft SQL Server to Snowflake using an updated_at column.

```yaml
name: raw.transactions
type: ingestr
parameters:
  source_connection: mssql_prod
  source_table: dbo.transactions
  destination: snowflake
materialization:
  type: table
  strategy: append
  incremental_key: updated_at
```

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

### Basic ingestr command structure

Source: https://getbruin.com/docs/ingestr/tutorials/load-kinesis-bigquery.html

The standard syntax for executing a data ingestion task using ingestr.

```bash
ingestr ingest \   
   --source-uri '<your-source-uri-here>' \   
   --source-table '<your-schema>.<your-table>' \   
   --dest-uri '<your-destination-uri-here>' \   
   --dest-table '<your-schema>.<your-table>'
```

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

### Create a dedicated replication user in Postgres

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-postgres-duckdb.html

Configures a role with necessary replication and select permissions, and creates a publication for the target table.

```sql
CREATE ROLE ingestr_cdc WITH LOGIN PASSWORD 'cdcpass' REPLICATION;
GRANT USAGE ON SCHEMA public TO ingestr_cdc;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ingestr_cdc;
CREATE PUBLICATION ingestr_pub FOR TABLE public.customers;
```

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

### Run assets by tag

Source: https://getbruin.com/docs/bruin/assets/definition-schema.html

Execute assets filtered by a specific tag using the Bruin CLI.

```bash
bruin run --tag client1
```

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

### Execute full refresh of an asset

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Command to perform an initial run of an SCD2 asset when the target table does not yet exist.

```bash
bruin run --full-refresh path/to/your/product_catalog.sql
```

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

### BigQuery destination URI configuration

Source: https://getbruin.com/docs/ingestr/tutorials/load-kinesis-bigquery.html

The URI format required to connect to a Google BigQuery project.

```bash
bigquery://<project-name>?credentials_path=/path/to/service/account.json&location=<location>
```

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

### Configure BigQuery Table Options

Source: https://getbruin.com/docs/bruin/assets/definition-schema.html

Defines partition-specific settings for BigQuery tables, such as expiration and filter requirements.

```yaml
materialization:
  type: table
  partition_by: event_date
bigquery:
  require_partition_filter: true
  partition_expiration_days: 30
  partition_key_immutable: true
```

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

### Execute incremental run of an asset

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Command to process new data and update the SCD2 table based on changes.

```bash
bruin run path/to/your/product_catalog.sql
```

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

### Perform full refresh

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-postgres-duckdb.html

Discards existing destination state and rebuilds the table from a fresh snapshot.

```bash
ingestr ingest \
  --source-uri "postgres+cdc://user:password@localhost:5432/shop?sslmode=disable" \
  --source-table "public.customers" \
  --dest-uri "duckdb:///shop.duckdb" \
  --dest-table "public.customers" \
  --full-refresh
```

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

### Perform a full refresh ingestion

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-sqlserver-duckdb.html

Use the --full-refresh flag to discard existing destination data and perform a fresh snapshot ingestion.

```bash
ingestr ingest \
  --source-uri "mssql+cdc://user:password@localhost:1433/shop?encrypt=disable" \
  --source-table "dbo.customers" \
  --dest-uri "duckdb:///warehouse.duckdb" \
  --dest-table "dbo.customers" \
  --full-refresh
```

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

### Configure full refresh restriction in environment

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Applies full refresh restriction to all assets within a production environment via the .bruin.yml file.

```yaml
environments:
  production:
    config:
      full_refresh_restricted: true
    connections:
      # ...
```

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

### Materialize Table with Append Strategy

Source: https://getbruin.com/docs/bruin/platforms/motherduck.html

Define a SQL asset to append new data to an existing table in MotherDuck.

```sql
/* @bruin
name: new_customers_per_region
type: duckdb.sql
connection: my_motherduck_connection
materialization:
    type: table
    strategy: append
@bruin */

SELECT COUNT(*) as customers, region 
FROM events.customers
    WHERE created_at >= {{ start_date }} 
      AND created_at < {{ end_date }}
```

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

### Open DuckDB database

Source: https://getbruin.com/docs/ingestr/tutorials/load-personio-duckdb.html

Access the DuckDB database file created during ingestion.

```bash
duckdb personio.duckdb
```

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

### Configure SCD2 with Incremental Key

Source: https://getbruin.com/docs/bruin/assets/materialization.html

YAML configuration for using a specific source column as the incremental timestamp key.

```yaml
materialization:
  type: table
  strategy: scd2_by_column
  incremental_key: updated_at
```

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

### Load an interval of data from Postgres to BigQuery

Source: https://getbruin.com/docs/ingestr/commands/ingest.html

Loads data within a specific date range using incremental keys and strategies.

```bash
ingestr ingest \
   --source-uri 'postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=disable' \
   --source-table 'public.users' \
   --dest-uri 'bigquery://my_project?credentials_path=/path/to/service/account.json&location=EU' \
   --dest-table 'raw.users' \
   --incremental-key 'dt' \
   --incremental-strategy 'delete+insert' \
   --interval-start '2023-01-01' \
   --interval-end '2023-01-31' \
   --columns 'dt:date'
```

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

### Time Interval Materialization Configuration

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Configures an asset for incremental updates based on a time interval and date key.

```yaml
materialization:
  type: table
  strategy: time_interval
  time_granularity: date
  incremental_key: dt

columns:
  - name: product_id
    type: INTEGER
    description: "Unique identifier for the product"
    primary_key: true
  - name: product_name
    type: VARCHAR
    description: "Name of the product"
  - name: price
    type: FLOAT
    description: "Price of the product in USD"
  - name: stock
    type: INTEGER
    description: "Number of units in stock"
  - name: dt
    type: DATE
    description: "Date when the product was last updated"
@bruin */


SELECT
    1 AS product_id,
    'Laptop' AS product_name,
    999.99 AS price,
    10 AS stock,
    DATE '2025-03-15' AS dt
UNION ALL
SELECT
    2 AS product_id,
    'Smartphone' AS product_name,
    699.99 AS price,
    50 AS stock,
    DATE '2024-03-16' AS dt;
```

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

### Define an Ingestr asset with a custom SQL query

Source: https://getbruin.com/docs/bruin/assets/ingestr.html

Use the query: prefix in the source_table parameter to execute custom SQL. Ensure the incremental key is included in the query results.

```yaml
name: raw.recent_orders
type: ingestr
parameters:
  source_connection: postgres_prod
  source_table: "query:select id, customer_id, updated_at from public.orders where updated_at > :interval_start"
  destination: bigquery
  incremental_strategy: merge
  incremental_key: updated_at

columns:
  - name: id
    type: integer
    primary_key: true
  - name: updated_at
    type: timestamp
```

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

### Define duckdb.sensor.query Asset

Source: https://getbruin.com/docs/bruin/platforms/motherduck.html

Configure a sensor to check for query results in MotherDuck.

```yaml
name: string
type: duckdb.sensor.query
connection: my_motherduck_connection
parameters:
    query: string
    timeout: duration (optional)
```

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

### Run ingestr with a specific publication

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-postgres-duckdb.html

Executes the ingestion process by pointing to the previously created publication via the source URI parameter.

```bash
ingestr ingest \
  --source-uri "postgres+cdc://ingestr_cdc:cdcpass@localhost:5432/shop?sslmode=disable&publication=ingestr_pub" \
  --source-table "public.customers" \
  --dest-uri "duckdb:///shop.duckdb" \
  --dest-table "public.customers"
```

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

### Create MySQL CDC User

Source: https://getbruin.com/docs/ingestr/tutorials/cdc-mysql-duckdb.html

Create a dedicated user with the necessary privileges for reading snapshots and streaming binary logs.

```sql
CREATE USER 'ingestr_cdc'@'%' IDENTIFIED BY 'cdcpass';
GRANT SELECT, RELOAD, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'ingestr_cdc'@'%';
FLUSH PRIVILEGES;
```

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

### Time Interval Materialization Execution

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Command to execute an asset with a specific time window for incremental loading.

```bash
bruin run --start-date "2024-03-01" --end-date "2024-03-31" path/to/your/asset
```

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

### Conceptual Merge Pattern

Source: https://getbruin.com/docs/ingestr/getting-started/incremental-loading.html

Illustrates the underlying SQL logic used by the merge strategy to handle staging and conditional updates.

```sql
CREATE OR REPLACE TABLE staging AS <rows read from source>;

MERGE INTO target AS t
USING (
  SELECT *
  FROM staging
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY primary_key
    ORDER BY incremental_key DESC -- where supported and provided
  ) = 1
) AS s
ON t.primary_key = s.primary_key
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT (...);
```

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

### Stream Data from Postgres to BigQuery

Source: https://getbruin.com/docs/ingestr/commands/ingest.html

Configures a continuous ingestion stream from a Postgres CDC source to BigQuery with custom flush thresholds.

```bash
# Stream all changes from a Postgres publication into BigQuery, flushing
# every 15 seconds or 100k changes, whichever comes first.
ingestr ingest \
   --source-uri 'postgres+cdc://user:pass@localhost:5432/mydb?publication=my_pub' \
   --dest-uri 'bigquery://my_project?credentials_path=/path/to/sa.json' \
   --stream \
   --flush-interval 15s \
   --flush-records 100000
```

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

### Run incremental update

Source: https://getbruin.com/docs/bruin/assets/materialization.html

Command to process new data updates for the SCD2 asset.

```bash
bruin run path/to/your/products.sql
```

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

### Load a specific query from Postgres to Snowflake

Source: https://getbruin.com/docs/ingestr/commands/ingest.html

Executes a custom SQL query with interval parameters to load data into Snowflake.

```bash
ingestr ingest \
   --source-uri 'postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=disable' \
   --dest-uri 'snowflake://user:password@account/dbname?warehouse=COMPUTE_WH&role=my_role' \
   --source-table 'query:SELECT * FROM public.users as pu JOIN public.orders as o ON pu.id = o.user_id WHERE pu.dt BETWEEN :interval_start AND :interval_end' \
   --dest-table 'raw.users' \
   --incremental-key 'dt' \
   --incremental-strategy 'delete+insert' \
   --interval-start '2023-01-01' \
   --interval-end '2023-01-31' \
   --columns 'dt:date'
```

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

### List DuckDB tables

Source: https://getbruin.com/docs/ingestr/tutorials/load-personio-duckdb.html

Check for the existence of tables within the DuckDB shell.

```sql
.tables;
```

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

### Configure asset execution status

Source: https://getbruin.com/docs/bruin/assets/definition-schema.html

Use the enabled property to toggle asset execution or provide a Jinja template for dynamic control.

```yaml
enabled: false
```

```yaml
enabled: "{{ var.asset_enabled }}"
```

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

### Copy a table from Postgres to DuckDB

Source: https://getbruin.com/docs/ingestr/commands/ingest.html

Basic command to migrate a table from a PostgreSQL source to a DuckDB destination.

```bash
ingestr ingest \
   --source-uri 'postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=disable' \
   --source-table 'public.input_table' \
   --dest-uri 'duckdb://output.duckdb' \
   --dest-table 'public.output_table'
```

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

### Ingest a CSV file to DuckDB

Source: https://getbruin.com/docs/ingestr/commands/ingest.html

Basic command to ingest data from a local CSV file into a DuckDB database.

```bash
ingestr ingest \
   --source-uri 'csv://input.csv' \
   --source-table 'sample' \
   --dest-uri 'duckdb://output.duckdb'
```

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

### Ingest data from source to destination

Source: https://getbruin.com/docs/ingestr/commands/ingest.html

Basic command structure for transferring data between a source and a destination.

```bash
ingestr ingest \
   --source-uri '<your-source-uri-here>' \
   --source-table '<your-schema>.<your-table>' \
   --dest-uri '<your-destination-uri-here>'
```

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

### Delete+insert logic pattern

Source: https://getbruin.com/docs/ingestr/getting-started/incremental-loading.html

Pseudocode illustrating the atomic transaction flow for the delete+insert strategy.

```sql
CREATE OR REPLACE TABLE staging AS <rows read from source>;

start_bound = COALESCE(--interval-start, MIN(staging.incremental_key));
end_bound   = COALESCE(--interval-end,   MAX(staging.incremental_key));

BEGIN TRANSACTION; -- or a destination-specific atomic/overwrite operation

DELETE FROM target
WHERE incremental_key >= start_bound
  AND incremental_key <= end_bound;

INSERT INTO target (columns...)
SELECT columns...
FROM staging;

COMMIT;
```

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

### Perform delete+insert incremental load

Source: https://getbruin.com/docs/ingestr/getting-started/incremental-loading.html

Stages rows from a source query and replaces the corresponding interval in the destination table based on the incremental key.

```bash
ingestr ingest \
    --source-uri 'postgresql://admin:admin@localhost:8837/web?sslmode=disable' \
    --source-table "query:SELECT * FROM my_schema.some_data WHERE dt = '2021-01-02'" \
    --dest-uri 'bigquery://<your-project-name>?credentials_path=/path/to/service/account.json' \
    --dest-table 'my_schema.some_data' \
    --incremental-strategy delete+insert \
    --incremental-key dt \
    --columns 'dt:date'
```

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

### Configure asset routing

Source: https://getbruin.com/docs/bruin/assets/definition-schema.html

Defines runtime routing options for execution environments that support named gateways.

```yaml
routing:
  egress_gateway: wg-shared-ams3
```

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

### Run ingestr for Stripe to PostgreSQL

Source: https://getbruin.com/docs/ingestr/tutorials/load-stripe-postgres.html

Specific command to execute the data transfer from Stripe to a PostgreSQL destination.

```bash
ingestr ingest \
    --source-uri 'stripe://?api_key=<YOUR_STRIPE_SECRET_KEY>' \
    --source-table 'event' \
    --dest-uri 'postgresql://<username>:<password>@<host>:<port>/<database-name>?sslmode=<sslmode>' \
    --dest-table 'dest.events'
```
Tokens:
Raw