Build an end to end data pipeline

Learn how to create an end-to-end data pipeline using Basin Pipelines, Basin Catalog, and Basin SQL for real-time transaction analysis.

Last updated Oct 12, 2026 In this tutorial, you will learn how to build a complete data pipeline using Basin Pipelines, Basin Catalog, and Basin SQL. This also includes a sample Python script that creates and sends financial transaction data to your Pipeline that can be queried by Basin SQL or any Apache Iceberg-compatible query engine.

This tutorial demonstrates how to:

  • Set up Basin Catalog to store our transaction events in an Apache Iceberg table
  • Set up a Basin Pipelines pipeline
  • Create transaction data with fraud patterns to send to your Pipeline
  • Query your data using Basin SQL for fraud analysis

Prerequisites

  1. Sign up for a Cloudflare account ↗︎.
  2. Install a Node.js version supported by Wrangler.
  3. Install Python 3.8+ ↗︎ for the data generation script.

Node.js version manager

Use Vite+ (vp env) ↗︎ to install and manage Node.js versions.

1. Set up authentication

You will need API tokens to interact with Cloudflare services.

  1. In the Cloudflare dashboard, go to the API tokens page. Go to Account API tokens ↗
  2. Select Create Token.
  3. Select Get started next to Create Custom Token.
  4. Enter a name for your API token.
  5. Under Permissions, choose:
  6. Basin Pipelines with Read, Send, and Edit permissions
  7. Basin Catalog with Read and Edit permissions
  8. Basin SQL with Read permissions
  9. Workers R2 Storage with Read and Edit permissions
  10. Optionally, add a TTL to this token.
  11. Select Continue to summary.
  12. Select Create Token.
  13. Note the Token value.

Export your new token as an environment variable:

export WRANGLER_BASIN_SQL_AUTH_TOKEN= #paste your token here

If this is your first time using Wrangler, make sure to log in.

npmyarnpnpm

npx wrangler login

yarn wrangler login

pnpm wrangler login

2. Create an R2 bucket and enable Basin Catalog

Create an R2 bucket:

npmyarnpnpm

npx wrangler r2 bucket create fraud-pipeline

yarn wrangler r2 bucket create fraud-pipeline

pnpm wrangler r2 bucket create fraud-pipeline
  1. In the Cloudflare dashboard, go to the R2 object storage page. Go to Overview ↗
  2. Select Create bucket.
  3. Enter the bucket name: fraud-pipeline
  4. Select Create bucket.

Enable the catalog on your R2 bucket:

npmyarnpnpm

npx wrangler basin catalog enable fraud-pipeline

yarn wrangler basin catalog enable fraud-pipeline

pnpm wrangler basin catalog enable fraud-pipeline

When you run this command, take note of the "Warehouse" and "Catalog URI". You will need these later.

  1. In the Cloudflare dashboard, go to the R2 object storage page. Go to Overview ↗
  2. Select the bucket: fraud-pipeline.
  3. Switch to the Settings tab, scroll down to Basin Catalog, and select Enable.
  4. Once enabled, note the Catalog URI and Warehouse name.

Note

Copy the warehouse (ACCOUNTID_BUCKETNAME) and paste it in the export below. We will use it later in the tutorial.

export WAREHOUSE= #Paste your warehouse here

(Optional) Enable compaction on your Basin Catalog

Basin Catalog can automatically compact tables for you. In production event streaming use cases, it is common to end up with many small files, so it is recommended to enable compaction. Since the tutorial only demonstrates a sample use case, this step is optional.

npmyarnpnpm

npx wrangler basin catalog compaction enable fraud-pipeline --token $WRANGLER_BASIN_SQL_AUTH_TOKEN

yarn wrangler basin catalog compaction enable fraud-pipeline --token $WRANGLER_BASIN_SQL_AUTH_TOKEN

pnpm wrangler basin catalog compaction enable fraud-pipeline --token $WRANGLER_BASIN_SQL_AUTH_TOKEN
  1. In the Cloudflare dashboard, go to the R2 object storage page. Go to Overview ↗
  2. Select the bucket: fraud-pipeline.
  3. Switch to the Settings tab, scroll to Basin Catalog, select the edit icon, and select Enable.
  4. Choose a target file size or leave the default, then select Save.

3. Set up the pipeline infrastructure

3.1. Create the Pipeline stream

First, create a schema file called raw_transactions_schema.json with the following json schema:

{
    "fields": [
        { "name": "transaction_id", "type": "string", "required": true },
        { "name": "user_id", "type": "int64", "required": true },
        { "name": "amount", "type": "float64", "required": false },
        { "name": "transaction_timestamp", "type": "string", "required": false },
        { "name": "location", "type": "string", "required": false },
        { "name": "merchant_category", "type": "string", "required": false },
        { "name": "is_fraud", "type": "bool", "required": false }
    ]
}

Create a stream to receive incoming fraud detection events:

npmyarnpnpm

npx wrangler basin pipelines streams create raw_events_stream --schema-file raw_transactions_schema.json --http-enabled true --http-auth false

yarn wrangler basin pipelines streams create raw_events_stream --schema-file raw_transactions_schema.json --http-enabled true --http-auth false

pnpm wrangler basin pipelines streams create raw_events_stream --schema-file raw_transactions_schema.json --http-enabled true --http-auth false

Note

Note the HTTP Ingest Endpoint URL from the output. This is the endpoint you will use to send data to your pipeline.

# The http ingest endpoint from the output (see example below)
export STREAM_ENDPOINT= #the http ingest endpoint from the output (see example below)

The output should look like this:

🌀 Creating stream 'raw_events_stream'...
✨ Successfully created stream 'raw_events_stream' with id 'stream_id'.

Creation Summary:
General:
  Name:  raw_events_stream

HTTP Ingest:
  Enabled:         Yes
  Authentication:  Yes
  Endpoint:        https://stream_id.ingest.cloudflare.com
  CORS Origins:    None

Input Schema:
┌───────────────────────┬────────┬────────────┬──────────┐
│ Field Name            │ Type   │ Unit/Items │ Required │
├───────────────────────┼────────┼────────────┼──────────┤
│ transaction_id        │ string │            │ Yes      │
├───────────────────────┼────────┼────────────┼──────────┤
│ user_id               │ int64  │            │ Yes      │
├───────────────────────┼────────┼────────────┼──────────┤
│ amount                │float64 │            │ No       │
├───────────────────────┼────────┼────────────┼──────────┤
│ transaction_timestamp │ string │            │ No       │
├───────────────────────┼────────┼────────────┼──────────┤
│ location              │ string │            │ No       │
├───────────────────────┼────────┼────────────┼──────────┤
│ merchant_category     │ string │            │ No       │
├───────────────────────┼────────┼────────────┼──────────┤
│ is_fraud              │ bool   │            │ No       │
└───────────────────────┴────────┴────────────┴──────────┘

3.2. Create the data sink

Create a sink that writes data to your R2 bucket as Apache Iceberg tables:

npmyarnpnpm

npx wrangler basin pipelines sinks create raw_events_sink --type "basin-catalog" --bucket "fraud-pipeline" --roll-interval 30 --namespace "fraud_detection" --table "transactions" --catalog-token $WRANGLER_BASIN_SQL_AUTH_TOKEN

yarn wrangler basin pipelines sinks create raw_events_sink --type "basin-catalog" --bucket "fraud-pipeline" --roll-interval 30 --namespace "fraud_detection" --table "transactions" --catalog-token $WRANGLER_BASIN_SQL_AUTH_TOKEN

pnpm wrangler basin pipelines sinks create raw_events_sink --type "basin-catalog" --bucket "fraud-pipeline" --roll-interval 30 --namespace "fraud_detection" --table "transactions" --catalog-token $WRANGLER_BASIN_SQL_AUTH_TOKEN

Note

This creates a sink configuration that will write to the Iceberg table fraud_detection.transactions in your Basin Catalog every 30 seconds. Basin Pipelines automatically appends an __ingest_ts column that is used to partition the table by DAY.

3.3. Create the pipeline

Connect your stream to your sink with SQL:

npmyarnpnpm

npx wrangler basin pipelines create raw_events_pipeline --sql "INSERT INTO raw_events_sink SELECT * FROM raw_events_stream"

yarn wrangler basin pipelines create raw_events_pipeline --sql "INSERT INTO raw_events_sink SELECT * FROM raw_events_stream"

pnpm wrangler basin pipelines create raw_events_pipeline --sql "INSERT INTO raw_events_sink SELECT * FROM raw_events_stream"
  1. In the Cloudflare dashboard, go to Basin Pipelines > Pipelines. Go to Pipelines ↗
  2. Select Create Pipeline.
  3. Connect to a Stream:
  4. Pipeline name: raw_events
  5. Enable HTTP endpoint for sending data: Enabled
  6. HTTP authentication: Disabled (default)
  7. Select Next
  8. Define Input Schema:
  9. Select JSON editor
  10. Copy in the schema:

json { "fields": [ { "name": "transaction_id", "type": "string", "required": true }, { "name": "user_id", "type": "int64", "required": true }, { "name": "amount", "type": "float64", "required": false }, { "name": "transaction_timestamp", "type": "string", "required": false }, { "name": "location", "type": "string", "required": false }, { "name": "merchant_category", "type": "string", "required": false }, { "name": "is_fraud", "type": "bool", "required": false } ] }

  • Select Next
  • Define Sink:
  • Select your R2 bucket: fraud-pipeline
  • Storage type: Basin Catalog
  • Namespace: fraud_detection
  • Table name: transactions
  • Advanced Settings: Change Maximum Time Interval to 30 seconds
  • Select Next
  • Credentials:
  • Disable Automatically create an Account API token for your sink
  • Enter Catalog Token from step 1
  • Select Next
  • Pipeline Definition:
  • Leave the default SQL query:

sql INSERT INTO raw_events_sink SELECT * FROM raw_events_stream;

  • Select Create Pipeline
  • After pipeline creation, note the Stream ID for the next step.

4. Generate sample fraud detection data

Create a Python script to generate realistic transaction data with fraud patterns:

fraud_data_generator.pypython

```python import requests import json import uuid import random import time import os from datetime import datetime, timezone, timedelta

Configuration - exported from the prior steps

STREAM_ENDPOINT = os.environ["STREAM_ENDPOINT"]# From the stream you created API_TOKEN = os.environ["WRANGLER_BASIN_SQL_AUTH_TOKEN"] #the same one created earlier EVENTS_TO_SEND = 1000 # Feel free to adjust this