Build an end to end data pipeline·续

def generate_transaction(): """Generate some random transactions with occasional fraud"""

User IDs

high_risk_users = [1001, 1002, 1003, 1004, 1005]
normal_users = list(range(1006, 2000))

user_id = random.choice(high_risk_users + normal_users) is_high_risk_user = user_id in high_risk_users

Generate amounts

if random.random() < 0.05:
    amount = round(random.uniform(5000, 50000), 2)
elif random.random() < 0.03:
    amount = round(random.uniform(0.01, 1.00), 2)
else:
    amount = round(random.uniform(10, 500), 2)

Locations

normal_locations = ["NEW_YORK", "LOS_ANGELES", "CHICAGO", "MIAMI", "SEATTLE", "SAN FRANCISCO"]
high_risk_locations = ["UNKNOWN_LOCATION", "VPN_EXIT", "MARS", "BAT_CAVE"]

if is_high_risk_user and random.random() < 0.3: location = random.choice(high_risk_locations) else: location = random.choice(normal_locations)

Merchant categories

normal_merchants = ["GROCERY", "GAS_STATION", "RESTAURANT", "RETAIL"]
high_risk_merchants = ["GAMBLING", "CRYPTO", "MONEY_TRANSFER", "GIFT_CARDS"]

if random.random() < 0.1: # 10% high-risk merchants merchant_category = random.choice(high_risk_merchants) else: merchant_category = random.choice(normal_merchants)

Series of checks to either increase fraud score by a certain margin

fraud_score = 0
if amount > 2000: fraud_score += 0.4
if amount < 1: fraud_score += 0.3
if location in high_risk_locations: fraud_score += 0.5
if merchant_category in high_risk_merchants: fraud_score += 0.3
if is_high_risk_user: fraud_score += 0.2

Compare the fraud scores

is_fraud = random.random() < min(fraud_score * 0.3, 0.8)

Generate timestamps (some fraud happens at unusual hours)

base_time = datetime.now(timezone.utc)
if is_fraud and random.random() < 0.4:  # 40% of fraud at night
    hour = random.randint(0, 5)  # Late night/early morning
    transaction_time = base_time.replace(hour=hour)
else:
    transaction_time = base_time - timedelta(
        hours=random.randint(0, 168)  # Last week
    )

return { "transaction_id": str(uuid.uuid4()), "user_id": user_id, "amount": amount, "transaction_timestamp": transaction_time.isoformat(), "location": location, "merchant_category": merchant_category, "is_fraud": True if is_fraud else False }

def send_batch_to_stream(events, batch_size=100): """Send events to Cloudflare Stream in batches"""

headers = { "Authorization": f"Bearer {API_TOKEN}", "Content-Type": "application/json" }

total_sent = 0 fraud_count = 0

for i in range(0, len(events), batch_size): batch = events[i:i + batch_size] fraud_in_batch = sum(1 for event in batch if event["is_fraud"] == True)

try: response = requests.post(STREAM_ENDPOINT, headers=headers, json=batch)

if response.status_code in [200, 201]: total_sent += len(batch) fraud_count += fraud_in_batch print(f"Sent batch of {len(batch)} events (Total: {total_sent})") else: print(f"Failed to send batch: {response.status_code} - {response.text}")

except Exception as e: print(f"Error sending batch: {e}")

time.sleep(0.1)

return total_sent, fraud_count

def main(): print("Generating fraud detection data...")

Generate events

events = []
for i in range(EVENTS_TO_SEND):
    events.append(generate_transaction())
    if (i + 1) % 100 == 0:
        print(f"Generated {i + 1} events...")

fraud_events = sum(1 for event in events if event["is_fraud"] == True) print(f"📊 Generated {len(events)} total events ({fraud_events} fraud, {fraud_events/len(events)*100:.1f}%)")

Send to stream

print("Sending data to Pipeline stream...")
sent, fraud_sent = send_batch_to_stream(events)

print(f"\nComplete!") print(f" Events sent: {sent:,}") print(f" Fraud events: {fraud_sent:,} ({fraud_sent/sent*100:.1f}%)") print(f" Data is now flowing through your pipeline!")

if name == "main": main()


Install the required Python dependency and run the script:

```bash
pip install requests
python fraud_data_generator.py

5. Query the data with Basin SQL

Now you can analyze your fraud detection data using Basin SQL. Here are some example queries:

5.1. View recent transactions

npmyarnpnpm

npx wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.transactions WHERE __ingest_ts > '2025-09-24T01:00:00Z' AND is_fraud = true LIMIT 10"

yarn wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.transactions WHERE __ingest_ts > '2025-09-24T01:00:00Z' AND is_fraud = true LIMIT 10"

pnpm wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.transactions WHERE __ingest_ts > '2025-09-24T01:00:00Z' AND is_fraud = true LIMIT 10"

5.2. Filter the raw transactions into a new table to highlight high-value transactions

Create a new sink that will write the filtered data to a new Apache Iceberg table in Basin Catalog:

npmyarnpnpm

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

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

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

Now you will create a new SQL query to process data from the original raw_events_stream stream and only write flagged transactions that are over the amount of 1,000.

npmyarnpnpm

npx wrangler basin pipelines create fraud_events_pipeline --sql "INSERT INTO fraud_filter_sink SELECT * FROM raw_events_stream WHERE is_fraud=true and amount > 1000"

yarn wrangler basin pipelines create fraud_events_pipeline --sql "INSERT INTO fraud_filter_sink SELECT * FROM raw_events_stream WHERE is_fraud=true and amount > 1000"

pnpm wrangler basin pipelines create fraud_events_pipeline --sql "INSERT INTO fraud_filter_sink SELECT * FROM raw_events_stream WHERE is_fraud=true and amount > 1000"

Note

It may take a few minutes for the new Pipeline to fully Initialize and start processing the data. Also keep in mind the 30 second roll-interval.

Query the table and check the results:

npmyarnpnpm

npx wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.fraud_transactions LIMIT 10"

yarn wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.fraud_transactions LIMIT 10"

pnpm wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.fraud_transactions LIMIT 10"

Also verify that the non-fraudulent events are being filtered out:

npmyarnpnpm

npx wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.fraud_transactions WHERE is_fraud = false LIMIT 10"

yarn wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.fraud_transactions WHERE is_fraud = false LIMIT 10"

pnpm wrangler basin sql query "$WAREHOUSE" "SELECT transaction_id, user_id, amount, location, merchant_category, is_fraud, transaction_timestamp FROM fraud_detection.fraud_transactions WHERE is_fraud = false LIMIT 10"

You should see the following output:

Query executed successfully with no results

Conclusion

You have successfully built an end to end data pipeline using Cloudflare's data platform. Through this tutorial, you have learned to:

  1. Use Basin Catalog: Leveraged Apache Iceberg tables for efficient data storage
  2. Set up Basin Pipelines: Created streams, sinks, and pipelines for data ingestion
  3. Generated sample data: Created transaction data with some basic fraud patterns
  4. Query your tables with Basin SQL: Access raw and processed data tables stored in Basin Catalog

Was this helpful?

YesNo

On this page

Build an end to end data pipeline·续 - 图1Docs

{"@context":"https://schema.org","@type":"TechArticle","@id":"https://developers.cloudflare.com/basin-sql/tutorials/end-to-end-pipeline/#page","headline":"Build an end to end data pipeline","description":"This tutorial demonstrates how to build a complete data pipeline using Basin Pipelines, Basin Catalog, and Basin SQL.","url":"https://developers.cloudflare.com/basin-sql/tutorials/end-to-end-pipeline/","inLanguage":"en","image":"https://developers.cloudflare.com/basin-sql/tutorials/end-to-end-pipeline/og.png?v=ab581c00f5892f2e","dateModified":"2026-10-12","publisher":{"@type":"Organization","name":"Cloudflare","description":"One platform for your apps, agents, and workforce. Build, secure, and scale without managing infrastructure","url":"https://www.cloudflare.com/","sameAs":["https://github.com/cloudflare","https://www.linkedin.com/company/cloudflare","https://x.com/cloudflare"],"logo":{"@type":"ImageObject","url":"https://developers.cloudflare.com/logo.svg"},"address":{"@type":"PostalAddress","streetAddress":"101 Townsend St","addressLocality":"San Francisco","addressRegion":"CA","postalCode":"94107","addressCountry":"US"},"contactPoint":[{"@type":"ContactPoint","contactType":"Customer Support","url":"https://support.cloudflare.com/","availableLanguage":["English"]},{"@type":"ContactPoint","contactType":"Sales","url":"https://www.cloudflare.com/contact/","availableLanguage":["English"]}]},"isPartOf":{"@type":"WebSite","@id":"https://developers.cloudflare.com/#website","name":"Cloudflare Docs","url":"https://developers.cloudflare.com/"}}

description: Basin SQL platform details including pricing and configuration. title: Platform image: https://developers.cloudflare.com/basin-sql/platform/og.png?v=448c59794ddf45a2

Skip to content

Documentation Index
Fetch the complete documentation index at: https://developers.cloudflare.com/basin-sql/llms.txt
Use this file to discover all available pages before exploring further.