Data pipeline com DLT, Dbt, Prefect e ClickHouse

Ultimamente venho estudando sobre Data Engineering e Data Pipelines. Achei interessante compartilhar um pouco do que ando aprendendo.

Onde quis fazer uma prova de conceito de uma pipeline de dados simples, mas que pudesse ser facilmente replicada em qualquer ambiente usando Docker, onde pudesse subir todo o ambiente usando Docker compose e que trouxesse conceitos de engenheria de software usando Dbt.

Além disso, um dos príncipios dessa arquitetura é custo, então pesquisei as que oferecerem a maior simplicidade e menor custo possível, assim quis usar o Prefect como orquestrador da pipeline, Dlt e o ClickHouse como data warehouse para armazenar os dados transformados.

Para tornar um pouco mais desafiador, quis fazer a ingestão de dados de banco MS SQL Server, que é um banco de dados que não costumo trabalhar no dia a dia, e o dataset seria sobre o táxi de Nova York, que é um dataset público e bem conhecido na comunidade de dados.

A arquitetura da solução ficou assim:

Criar o ambiente com Docker Compose

Primeiro passo é criar o ambiente com Docker Compose, onde vou subir o banco de dados de origem. Criei o arquivo docker-compose.yml na pasta raiz do seu projeto com o seguinte conteúdo:

1
services:
2
  sqlserver:
3
    image: mcr.microsoft.com/mssql/server:2022-latest
4
    container_name: sqlserver
5
    environment:
6
      ACCEPT_EULA: 'Y'
7
      MSSQL_SA_PASSWORD: 'YourStrong!Passw0rd'
8
    ports:
9
      - '1433:1433'
10
    volumes:
11
      - sqlserverdata:/var/opt/mssql

Em seguida inicializo o container:

1
docker-compose up -d

Importar os dados no SQL Server

Fiz o download dos dados do NYC Taxi & Limousine Commission (TLC) Trip Record Data através desse link, e salvei o arquivo na pasta dataset do meu projeto.

Para importar o arquivo de backup .bak para dentro do container do SQL Server, uso o seguinte comando:

1
docker cp ./dataset/NYCTaxi_Sample.bak sqlserver:/var/opt/mssql/data/NYCTaxi_Sample.bak

Em seguida, executo o comando de restauração do banco de dados:

1
docker exec -it sqlserver /opt/mssql-tools18/bin/sqlcmd \
2
   -S localhost -U sa -P 'YourStrong!Passw0rd' -C \
3
   -Q 'RESTORE DATABASE NYCTaxi_Sample FROM DISK = "/var/opt/mssql/data/NYCTaxi_Sample.bak" WITH MOVE "NYCTaxi_Sample" TO "/var/opt/mssql/data/NYCTaxi_Sample.mdf", MOVE "NYCTaxi_Sample_log" TO "/var/opt/mssql/data/NYCTaxi_Sample_log.ldf"'

Criar estrutura do projeto

Agora que o banco de dados está pronto, vou criar a estrutura do projeto com as ferramentas que vou utilizar: DLT, ClickHouse, Dbt e Prefect.

Primeiro organizo a estrutura de pastas do projeto:

1
data-engineer/
2
├── dataset/            # Raw data backups
3
├── nyc_taxi/
4
│   ├── main_flow.py    # Prefect orchestrator
5
│   ├── extract_sqlserver.py  # DLT ingestion logic
6
│   └── nyc_taxi_dbt/   # dbt project
7
│       ├── models/     # SQL Transformation models
8
│       └── profiles.yml # dbt connection settings
9
└── docker-compose.yaml # Infrastructure definition

Dentro da pasta nyc_taxi, crio o ambiente virtual do Python e instalo as dependências necessárias usando UV. O UV é uma ferramenta de gerenciamento de dependências e ambientes virtuais para projetos Python construída em Rust (recomendo você conhecer).

1
brew install uv

Em seguida, inicializo o ambiente UV:

1
uv init nyc_taxi

Adiciono as dependências do projeto:

1
uv add dbt-core dbt-sqlserver dbt-clickhouse prefect prefect-client

Em seguida, criar o projeto Dbt:

Configurar o Dbt

O Dbt (Data Build Tool) é uma ferramenta de transformação de dados que permite aos engenheiros de dados e analistas transformar, testar e documentar dados em seus armazéns de dados. Para criar o projeto Dbt, navego até a pasta nyc_taxi e executo o comando:

1
cd nyc_taxi
2
dbt init nyc_taxi_dbt

Meu arquivo dbt_project.yml ficou assim:

1
# Name your project! Project names should contain only lowercase characters
2
# and underscores. A good package name should reflect your organization's
3
# name or the intended use of these models
4
name: 'nyc_taxi_dbt'
5
version: '1.0.0'
6

7
# This setting configures which "profile" dbt uses for this project.
8
profile: 'nyc_taxi_clickhouse'
9

10
# These configurations specify where dbt should look for different types of files.
11
# The `model-paths` config, for example, states that models in this project can be
12
# found in the "models/" directory. You probably won't need to change these!
13
model-paths: ['models']
14
analysis-paths: ['analyses']
15
test-paths: ['tests']
16
seed-paths: ['seeds']
17
macro-paths: ['macros']
18
snapshot-paths: ['snapshots']
19

20
clean-targets: # directories to be removed by `dbt clean`
21
  - 'target'
22
  - 'dbt_packages'
23

24
# Configuring models
25
# Full documentation: https://docs.getdbt.com/docs/configuring-models
26

27
# In this example config, we tell dbt to build all models in the example/
28
# directory as views. These settings can be overridden in the individual model
29
# files using the `{{ config(...) }}` macro.
30
models:
31
  nyc_taxi_dbt:
32
    # Config indicated by + and applies to all files under models/example/
33
    example:
34
      +materialized: view

Configuro o arquivo ~/.dbt/profiles.yml para conectar no ClickHouse:

1
nyc_taxi_clickhouse:
2
  target: dev
3
  outputs:
4
    dev:
5
      type: clickhouse
6
      host: localhost
7
      port: 8123
8
      user: default
9
      password: password
10
      schema: nyc_taxi
11
      threads: 4

Vou pode testar se o projeto Dbt está funcionando corretamente executando o comando:

1
uv run dbt debug

Configurar os Dbt models

Vou criar os modelos Dbt para transformar os dados. Crio as pastas e arquivos necessários dentro da pasta nyc_taxi/nyc_taxi_dbt/models/:

1
data-engineer/
2
├── nyc_taxi/
3
│   └── nyc_taxi_dbt/   # dbt project
4
│       └── models/     # SQL Transformation models
5
│           ├── staging/
6
│           │   ├── sources.yml
7
│           │   └── stg_nyctaxi_sample.sql
8
│           └── marts/
9
│               └── fact_nyctaxi_trips.sql

O arquivo sources.yml define a fonte de dados:

1
version: 2
2

3
sources:
4
  - name: clickhouse_staging
5
    database: nyc_taxi
6
    tables:
7
      - name: nyctaxi_sample
8
        identifier: nyc_taxi_staging___nyctaxi_sample

O arquivo stg_nyctaxi_sample.sql cria a tabela de staging:

1
{{ config(materialized='view') }}
2

3
with source as (
4
    select * from {{ source('clickhouse_staging', 'nyctaxi_sample') }}
5
),
6

7
renamed as (
8
    select
9
        medallion,
10
        hack_license,
11
        vendor_id,
12
        rate_code,
13
        store_and_fwd_flag,
14
        pickup_datetime,
15
        dropoff_datetime,
16
        passenger_count,
17
        trip_time_in_secs,
18
        trip_distance,
19
        pickup_longitude,
20
        pickup_latitude,
21
        dropoff_longitude,
22
        dropoff_latitude,
23
        payment_type,
24
        fare_amount,
25
        surcharge,
26
        mta_tax,
27
        tolls_amount,
28
        total_amount,
29
        tip_amount,
30
        tipped,
31
        tip_class,
32
        _dlt_load_id,
33
        _dlt_id
34
    from source
35
)
36

37
select * from renamed

O arquivo fact_nyctaxi_trips.sql cria a tabela fato:

1
{{ config(
2
    materialized='incremental',
3
    engine='ReplacingMergeTree',
4
    order_by=['medallion', 'hack_license', 'pickup_datetime'],
5
    unique_key='_dlt_id',
6
    incremental_strategy='append'
7
) }}
8

9
-- ReplacingMergeTree handles duplicates automatically based on the ORDER BY keys
10
-- when merges happen. In dbt-clickhouse, 'incremental' with 'append' is often used
11
-- with ReplacingMergeTree to let the engine handle deduplication.
12

13
with staging as (
14
    select * from {{ ref('stg_nyctaxi_sample') }}
15
    {% if is_incremental() %}
16
    where pickup_datetime > (select max(pickup_datetime) from {{ this }})
17
    {% endif %}
18
)
19

20
select * from staging

A razão de usar o ReplacingMergeTree é que ele permite que o ClickHouse gerencie automaticamente a substituição de registros duplicados com base na chave primária definida (neste caso, medallion, hack_license e pickup_datetime). Isso é especialmente útil para cenários de carga incremental, onde novos dados podem conter atualizações ou correções para registros existentes.

Além disso dividi em staging e marts para seguir as boas práticas de organização de projetos Dbt, onde os dados brutos são primeiro carregados em tabelas de staging antes de serem transformados em tabelas fato ou dimensão.

Se você quiser mais sobre estratégias de modelagem como dimensões e fatos, você pode ler esse artigo.

Então, ilustrando o fluxo de dados, seria:

1
[Raw Data in ClickHouse] --> [StagingTable: stg_nyctaxi_sample] --> [FactTable: fact_nyctaxi_trips]

Com isso finalizo a configuração do Dbt. Podemos testar os modelos executando o comando:

1
uv run dbt run

Configurar o Docker Compose para ClickHouse e Prefect

Agora vou adicionar os serviços do ClickHouse e do Prefect no arquivo docker-compose.yml:

1
services:
2
  sqlserver: # Abaixo o serviço do SQL Server já criado
3
  clickhouse:
4
    image: clickhouse/clickhouse-server
5
    container_name: clickhouse
6
    environment:
7
      CLICKHOUSE_USER: default
8
      CLICKHOUSE_PASSWORD: password
9
      CLICKHOUSE_DB: nyc_taxi
10
      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
11
    ports:
12
      - '8123:8123'
13
      - '9000:9000'
14
    volumes:
15
      - clickhousedata:/var/lib/clickhouse
16
  prefect:
17
    image: prefecthq/prefect:3-python3.12
18
    container_name: prefect
19
    environment:
20
      PREFECT_SERVER_API_HOST: 0.0.0.0
21
    command: prefect server start --no-services
22
    ports:
23
      - '4200:4200'
24
    volumes:
25
      - prefectdata:/var/lib/prefect
26
volumes:
27
  sqlserverdata:
28
  clickhousedata:
29
  prefectdata:

Vamos subir os containers do ClickHouse e do Prefect:

1
docker-compose up -d clickhouse prefect

Criar o fluxo de orquestração com Prefect

Agora vou criar o fluxo de orquestração usando o Prefect. Crio o arquivo main_flow.py dentro da pasta nyc_taxi com o seguinte conteúdo:

1
import logging
2
from prefect import flow, task
3
from extract_sqlserver import load_sql_server_to_clickhouse
4
from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings
5

6
# Configure logging
7
logging.basicConfig(level=logging.INFO)
8
logger = logging.getLogger(__name__)
9

10
@task(retries=3, retry_delay_seconds=60)
11
def extract_task():
12
    logger.info("Starting extraction task...")
13
    load_sql_server_to_clickhouse()
14
    logger.info("Extraction task completed.")
15

16
@task
17
def dbt_run_task():
18
    logger.info("Starting dbt run task...")
19
    result = PrefectDbtRunner(
20
        settings=PrefectDbtSettings(
21
            project_dir="nyc_taxi_dbt",
22
            profiles_dir="nyc_taxi_dbt"
23
        )
24
    ).invoke(["build"])
25
    logger.info("dbt run task completed.")
26

27
@task
28
def data_quality_checks():
29
    logger.info("Running data quality checks...")
30
    # This could be more dbt tests or custom SQL checks
31
    # For now, we'll assume dbt build (which includes tests) covers this.
32
    logger.info("Data quality checks passed.")
33

34
@flow(name="nyc_taxi_etl")
35
def nyc_taxi_pipeline():
36
    extract_task()
37
    dbt_run_task()
38
    data_quality_checks()
39
    # Alerting can be handled by Prefect's native automation or state handlers
40
    logger.info("Pipeline completed successfully.")
41

42
if __name__ == "__main__":
43
    nyc_taxi_pipeline.serve(name="nyc_taxi_pipeline", cron="0 */12 * * *")

Em seguida, crio o arquivo extract_sqlserver.py com a lógica de extração dos dados do SQL Server para o ClickHouse:

1
import logging
2
import dlt
3
from dlt.sources.sql_database import sql_database
4

5
def load_sql_server_to_clickhouse():
6
    # Configure logging to see what's happening
7
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
8
    logger = logging.getLogger(__name__)
9

10
    logger.info("Starting SQL Server to ClickHouse extraction...")
11
    # Configure the destination: ClickHouse
12
    # dlt will automatically pick up credentials from .dlt/secrets.toml
13
    pipeline = dlt.pipeline(
14
        pipeline_name="sql_server_to_clickhouse",
15
        destination="clickhouse"
16
    )
17

18
    # Define the source with incremental logic
19
    # dlt will automatically pick up credentials for sql_database from .dlt/secrets.toml
20
    source = sql_database().with_resources("nyctaxi_sample")
21

22
    # Configure incremental loading and primary keys for the resource
23
    # To use ReplacingMergeTree in ClickHouse, we need a primary key.
24
    # medallion + hack_license + pickup_datetime seem like a good candidate for a unique key in this sample.
25
    source.nyctaxi_sample.apply_hints(
26
        incremental=dlt.sources.incremental("pickup_datetime"),
27
        primary_key=["medallion", "hack_license", "pickup_datetime"]
28
    )
29

30
    # Run the pipeline
31
    # Use 'merge' to support updates/CDC and trigger ReplacingMergeTree.
32
    info = pipeline.run(source, write_disposition="merge")
33

34
    print(info)
35

36
if __name__ == "__main__":
37
    load_sql_server_to_clickhouse()

Com isso vamos executar o fluxo de orquestração dentro da pasta nyc_taxi onde contém o ambiente virtual do UV:

1
cd nyc_taxi
2
uv run python main_flow.py

Você pode monitorar a execução do fluxo acessando o dashboard do Prefect em http://localhost:4200.

Nos logs do fluxo você verá as etapas de extração, transformação e carregamento dos dados:

Você pode verificar os dados carregados no ClickHouse usando o cliente web em http://localhost:8123 ou qualquer ferramenta de consulta SQL compatível com ClickHouse.

Aqui você pode ver a tabela fato fact_nyctaxi_trips criada no ClickHouse:

Configurar Clickhouse UI

Para facilitar a visualização dos dados no ClickHouse, você pode usar o ClickHouse UI.

Vamos adicionar o serviço do ClickHouse UI no arquivo docker-compose.yml:

1
  ch-ui:
2
    image: ghcr.io/caioricciuti/ch-ui:latest
3
    restart: always
4
    ports:
5
      - '5521:5521'
6
    environment:
7
      # Core ClickHouse Configuration
8
      VITE_CLICKHOUSE_URL: 'http://localhost:8123'
9
      VITE_CLICKHOUSE_USER: 'default'
10
      VITE_CLICKHOUSE_PASS: 'password'
11

12
      # Optional: Advanced Features
13
      VITE_CLICKHOUSE_USE_ADVANCED: 'false'
14
      VITE_CLICKHOUSE_CUSTOM_PATH: ''
15
      VITE_CLICKHOUSE_REQUEST_TIMEOUT: '30000'
16

17
      # Optional: Reverse Proxy Support
18
      VITE_BASE_PATH: '/'

Em seguida, subo o container do ClickHouse UI:

1
docker-compose up -d ch-ui

Você pode acessar a interface do ClickHouse UI em http://localhost:5521 para explorar os dados carregados. Na imagem abaixo você pode ver que fiz uma consulta na tabela fato fact_nyctaxi_trips para contar todas as viagens registradas e levou apenas 1.18ms onde temo 1.703.957 registros carregados. É muito rápido!:

Conclusão

É muito gratificante ver como todas essas ferramentas podem trabalhar juntas para criar uma pipeline de dados eficiente e escalável. O uso do DLT para ingestão, Dbt para transformação, Prefect para orquestração e ClickHouse como data warehouse proporciona uma solução robusta e de alto desempenho, que pode ser facilmente replicada em diferentes ambientes graças ao Docker Compose.

Existem outras ferramentas de orquestração como Dagster ou Airflow, mas escolhi o Prefect por sua simplicidade e facilidade de uso, além de sua execução ser realizada no próprio code base.

Além disso, para ingestão de dados, existe o Airbyte, mas escolhi o DLT, pois oferece uma abordagem moderna, eficiente e utiliza Apache Arrow, especialmente quando combinado com o ClickHouse, que é conhecido por sua velocidade e capacidade de lidar com grandes volumes de dados.

Em relação ao Dbt, eu poderia ter usado o Dbt fusion, que é uma nova versão desenvolvida em Rust e que utiliza Apache Arrow, mas ainda está em fase beta, então optei pelo Dbt tradicional.

Apache Arrow é uma tecnologia poderosa que está ganhando cada vez mais espaço no ecossistema de dados, e é interessante ver como ela está sendo integrada em várias ferramentas modernas de engenharia de dados. The

O formato Arrow é uma maneira melhor de representar dados tabulares na memória do que objetos nativos do Python (lista de dicionários). Ele permite descarregar o processamento para a biblioteca C++ rápida do Arrow e evita o processamento de linhas uma a uma. Se você se interessa em entender mais sobre, recomendo esse vídeo.

Referências:

• NYC Taxi demo data for SQL Server Python and R tutorials
• Code along - build an ELT Pipeline in 1 Hour (dbt, Snowflake, Airflow)
• Dbt docs
• Dlthub docs
• Moving away from Segment to a cost-effective do-it-yourself event streaming pipeline with Cloud Pub/Sub and dlt.
• The Hidden Costs of Running Apache Airflow
• Airflow Local Development Sucks
• Prefect docs
• ClickHouse docs