Skip to main content

The Starrocks adapter plugin for dbt

Project description

dbt-starrocks-droppii

Fork notice: This is a fork of StarRocks/dbt-starrocks, maintained by Droppii and published as dbt-starrocks-droppii. See CHANGELOG.md for fork-specific fixes.

PyPI PyPI - Python Version PyPI - Downloads

This project is under development.

The dbt-starrocks-droppii package contains all the code to enable dbt to work with StarRocks.

  • Requires StarRocks version 2.5.0 or higher
    • version 3.4.x is recommended
    • StarRocks versions 2.4 and below are no longer supported

Installation

This plugin can be installed via pip:

$ pip install dbt-starrocks-droppii

Supported features

StarRocks <= 2.5 StarRocks 2.5 ~ 3.1 StarRocks >= 3.1 StarRocks >= 3.4 Feature
Table materialization
View materialization
Materialized View materialization
Incremental materialization
Primary Key Model
Sources
Custom data tests
Docs generate
Expression Partition
Kafka
Dynamic Overwrite
*4 *4 Submit task
Microbatch (Insert Overwrite)
Microbatch (Dynamic Overwrite)

Notice

  1. When StarRocks Version < 2.5, Create table as can only set engine='OLAP' and table_type='DUPLICATE'
  2. When StarRocks Version >= 2.5, Create table as supports table_type='PRIMARY'
  3. When StarRocks Version < 3.1 distributed_by is required
  4. Verify the specific submit task support for your version, see SUBMIT TASK.
  5. Views: when a view's SQL is unchanged, dbt run issues no DDL on the view, leaving it in place (the run log notes skip <view>, and the model still completes as a successful no-op). This avoids deactivating dependent materialized views, which StarRocks does whenever a base view is recreated, even with identical SQL.

Profile Configuration

Example entry for profiles.yml:

starrocks:
  target: dev
  outputs:
    dev:
      type: starrocks
      host: localhost
      port: 9030
      catalog: default_catalog
      schema: analytics
      username: your_starrocks_username
      password: your_starrocks_password
Option Description Required? Example
type The specific adapter to use Required starrocks
host The hostname to connect to Required 192.168.100.28
port The port to use Required 9030
catalog Specify the catalog to build models into Optional default_catalog
schema Specify the schema (database in StarRocks) to build models into Required analytics
username The username to use to connect to the server Required dbt_admin
password The password to use for authenticating to the server Required correct-horse-battery-staple
version Let Plugin try to go to a compatible starrocks version Optional 3.1.0
use_pure set to "true" to use C extensions Optional true
is_async "true" to submit suitable tasks as etl tasks. Optional true
async_query_timeout Sets the query_timeout value when submitting a task to StarRocks Optional 300
poll_interval Base delay in seconds between task status polls Optional 1
poll_max_delay Maximum delay cap in seconds for task polling Optional 600
poll_factor Growth multiplier for exponential backoff between polls Optional 2.0

More details about setting use_pure and other connection arguments here

Example

dbt seed properties(yml):

Complete configuration:

models:
  materialized: table                   // table, view, materialized_view or incremental
  engine: 'OLAP'
  keys: ['id', 'name', 'some_date']
  table_type: 'PRIMARY'                 // PRIMARY or DUPLICATE or UNIQUE
  distributed_by: ['id']
  buckets: 3                            // leave empty for auto bucketing
  indexs=[{ 'columns': 'idx_column' }]
  partition_by: ['some_date']
  partition_by_init: ["PARTITION p1 VALUES [('1971-01-01 00:00:00'), ('1991-01-01 00:00:00')),PARTITION p1972 VALUES [('1991-01-01 00:00:00'), ('1999-01-01 00:00:00'))"]
  // RANGE, LIST, or Expr partition types should be used in conjunction with partition_by configuration
  // Expr partition type requires an expression (e.g., date_trunc) specified in partition_by
  order_by: ['some_column']             // only for PRIMARY table_type
  partition_type: 'RANGE'               // RANGE or LIST or Expr Need to be used in combination with partition_by configuration
  properties: {"replication_num":"1", "in_memory": "true"}
  refresh_method: 'async'               // only for materialized view default manual
  on_view_exists: 'replace'             // only for view: use CREATE OR REPLACE VIEW instead of DROP + CREATE

  // For 'materialized=incremental' in version >= 3.4
  incremental_strategy: 'dynamic_overwrite' // Supported values: ['default', 'insert_overwrite', 'dynamic_overwrite']

  // For 'materialized=incremental' and 'incremental_strategy=microbatch'
  event_time: 'some_timestamp_column'     // The column name of the event time
  begin: '2025-01-01'                     // The start time of the incremental data
  lookback: 1                             // The lookback time of the each incremental run
  batch_size: 'day'                       // The batch size. Supported values ['year', 'month', 'day', 'hour']
  microbatch_use_dynamic_overwrite: true  // Whether to use dynamic_overwrite in version >= 3.4

dbt run config:

Example configuration:

{{ config(materialized='view') }}
{{ config(materialized='view', on_view_exists='replace') }}
{{ config(materialized='table', engine='OLAP', buckets=32, distributed_by=['id']) }}
{{ config(materialized='table', indexs=[{ 'columns': 'idx_column' }]) }}
{{ config(materialized='table', partition_by=['date_trunc("day", first_order)'], partition_type='Expr') }}
{{ config(materialized='table', table_type='PRIMARY', keys=['customer_id'], order_by=['first_name', 'last_name'] }}
{{ config(materialized='incremental', table_type='PRIMARY', engine='OLAP', buckets=32, distributed_by=['id']) }}
{{ config(materialized='incremental', partition_by=['my_partition_key'], partition_type='Expr', incremental_strategy='dynamic_overwrite') }}
{{ config(materialized='incremental', partition_by=['my_partition_key'], partition_type='Expr', incremental_strategy='microbatch', event_time='report_day', begin='2025-01-01', lookback=1, batch_size='day') }}
{{ config(materialized='incremental', partition_by=['my_partition_key'], partition_type='Expr', incremental_strategy='microbatch', event_time='report_day', begin='2025-01-01', lookback=1, batch_size='day', microbatch_use_dynamic_overwrite=true) }}
{{ config(materialized='materialized_view') }}
{{ config(materialized='materialized_view', properties={"storage_medium":"SSD"}) }}
{{ config(materialized='materialized_view', refresh_method="ASYNC START('2022-09-01 10:00:00') EVERY (interval 1 day)") }}

For materialized view only support partition_by、buckets、distributed_by、properties、refresh_method configuration.

Read From Catalog

First you need to add this catalog to starrocks. The following is an example of hive.

CREATE EXTERNAL CATALOG `hive_catalog`
PROPERTIES (
    "hive.metastore.uris"  =  "thrift://127.0.0.1:8087",
    "type"="hive"
);

How to add other types of catalogs can be found in the documentation. https://docs.starrocks.io/en-us/latest/data_source/catalog/catalog_overview Then write the sources.yaml file.

sources:
  - name: external_example
    schema: hive_catalog.hive_db
    tables:
      - name: hive_table_name

Finally, you might use below marco quote

{{ source('external_example', 'hive_table_name') }}

Dynamic Overwrite (StarRocks >= 3.4)

Add a new incremental_strategy property that supports the following values:

  • default (or omitted): Standard inserts without overwrite.
  • insert_overwrite: Will apply overwrite with dynamic_overwrite = false to the inserts.
  • dynamic_overwrite: Will apply overwrite with dynamic_overwrite = true to the inserts.

For more details on the different behaviors, see StarRocks' documentation for INSERT.

Submittable ETL tasks

The implementation of the submittable etl is located in the impl.py file.

Setting is_async: true in your profiles.yml will enable submitting suitable ETL tasks using the submit task feature of StarRocks.

This will be automatically wrapped around any statement that supports submission. Setting this manually is currently not supported by the adapter.

The following statements will be submitted automatically:

  • CREATE AS ... SELECT
  • INSERT INTO|OVERWRITE
  • CACHE SELECT ...

See StarRocks' documentation on SUBMIT TASK

Task Polling

Once the task has been submitted, the adapter will periodically poll StarRocks' information_schema.task_runs to retrieve the task status.

The polling is implemented using a configurable exponential backoff. The adapter's connection to the StarRocks' cluster will not be maintained during the waiting period. It will be re-opened right before the next status polling phase.

The polling delay is calculated as: min(poll_max_delay, poll_interval * (poll_factor ^ attempt))

Option Description Default
poll_interval Base delay in seconds between polls 1
poll_max_delay Maximum delay cap in seconds 600
poll_factor Growth multiplier for exponential backoff 2.0

For example, with poll_interval: 5, poll_factor: 1.5, poll_max_delay: 60:

Attempt Delay
1 7.5s
2 11.25s
3 16.9s
4 25.3s
5 38s
6 57s
7+ 60s (capped)

Compared to defaults (poll_interval: 1, poll_factor: 2.0, poll_max_delay: 600):

Attempt Delay
1 2s
2 4s
3 8s
4 16s
5 32s
6 64s
7 128s
8 256s
9 512s
10+ 600s (capped)

Controlling the task timeout

Using the async_query_timeout property in the profiles.yml will control the value of the query_timeout when submitting task.

It's going to be injected in the SQL query submitted to StarRocks:

submit /*+set_var(query_timeout={async_query_timeout})*/ task ...

Example profiles.yml configuration

my_profile:
  target: dev
  outputs:
    dev:
      type: starrocks
      host: host
      port: 9030
      schema: schema
      username: username
      password: password
      is_async: true
      async_query_timeout: 3600 # 1 hour
      poll_interval: 5
      poll_max_delay: 60
      poll_factor: 1.5

Test Adapter

Prepare env: install dev requirements and run StarRocks docker image

pip install -r dev_requirements.txt
docker run -p 9030:9030 starrocks/allin1-ubuntu:3.5-latest

Run the following

python3 -m pytest tests/functional

consult the project

Contributing

We welcome you to contribute to dbt-starrocks. Please see the Contributing Guide for more information.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

dbt_starrocks_droppii-1.12.2.tar.gz (35.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

dbt_starrocks_droppii-1.12.2-py3-none-any.whl (48.8 kB view details)

Uploaded Python 3

File details

Details for the file dbt_starrocks_droppii-1.12.2.tar.gz.

File metadata

  • Download URL: dbt_starrocks_droppii-1.12.2.tar.gz
  • Upload date:
  • Size: 35.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for dbt_starrocks_droppii-1.12.2.tar.gz
Algorithm Hash digest
SHA256 e56ab235e7170991ac77f40f8f9bc4ef2e108b1ec849c10d52e7942a3afd6766
MD5 6066286e7b6a8dc6681d7fdc242664d0
BLAKE2b-256 f85b2b38a6c64e63331ecf4eb2b286075f7ec2943043a119c622eaf273dc3057

See more details on using hashes here.

File details

Details for the file dbt_starrocks_droppii-1.12.2-py3-none-any.whl.

File metadata

File hashes

Hashes for dbt_starrocks_droppii-1.12.2-py3-none-any.whl
Algorithm Hash digest
SHA256 9fc39a3a9e347164436c9e5ff117deabf68034491cc304089a3bb734fc041d46
MD5 d23303702b527d43c5437f8975fd4117
BLAKE2b-256 7091f17bb228164584c04fecc1be58f4229bae4fbf4f14be928d63b8617f5b54

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page