DEV Community: pawan deore

Apache Spark Partitioning, Skew, and Broadcast Joins: A Practical Guide

pawan deore — Sun, 08 Nov 2026 12:15:20 +0000

Apache Spark Partitioning, Skew, and Broadcast Joins: A Practical Guide

Apache Spark can process large datasets in parallel because it divides work into partitions. When those partitions are balanced, executors finish work at roughly the same time. When one partition contains far more data than the others, a single task becomes a bottleneck. This is called data skew.

Partitioning, skew handling, and join strategy are tightly connected:

Partitioning determines where records are processed.
Joins often trigger expensive data shuffles.
Skew causes a small number of join or aggregation tasks to run much longer than the rest.
Broadcast joins can avoid shuffling a large fact table when one side is small enough.

This tutorial assumes Spark DataFrames, PySpark, and a typical analytics workload with a large fact table and smaller lookup tables.

Partitioning: Spark’s Unit of Parallel Work

A Spark DataFrame is split into partitions. Each partition is processed by one task at a time, typically on an executor core.

For example, a DataFrame with 200 partitions can create up to 200 parallel tasks for a stage. The actual concurrency depends on available executor cores.

There are several different meanings of “partition” in data engineering, and confusing them causes design mistakes:

Term	Meaning
Spark partition	An in-memory or execution-time chunk of a DataFrame or RDD
Shuffle partition	A partition created after Spark redistributes data for joins, aggregations, or `repartition()`
File partition	A physical output directory such as `date=2025-03-08/` in Parquet storage
Table partition	A logical storage partition defined in a metastore or table format

A Parquet table partitioned by event_date can still have poorly balanced Spark partitions during a join. Storage layout and execution layout are related, but they are not the same thing.

The most important operations are:

df.repartition(200)
df.repartition(200, "customer_id")
df.repartitionByRange(200, "event_date")
df.coalesce(50)

repartition() creates a full shuffle. Spark moves rows across executors to produce the requested partitioning.

coalesce() usually reduces partitions without a full shuffle. It is useful after filtering data down, but it does not rebalance uneven partitions. If one partition was already huge, coalesce() can preserve that imbalance.

How Joins Create Shuffles

Consider a common pipeline:

Raw order files
    |
    v
Filter and clean orders
    |
    +--------------------+
    |                    |
    v                    v
Large orders table   Small customers table
    |                    |
    +------ join --------+
              |
              v
Aggregate revenue by customer segment
              |
              v
Write Parquet or Delta output

A join between two large DataFrames usually requires Spark to shuffle both sides by the join key:

orders                          customers
  |                                  |
  v                                  v
Exchange hash(customer_id)      Exchange hash(customer_id)
  |                                  |
  +------------- join -------------+

If both tables are distributed by customer_id, rows for the same customer are placed in the same shuffle partition. That makes the join possible, but it creates a problem when one customer appears far more often than others.

For example:

customer_id = C_HOT      700 million orders
customer_id = C_001        3 million orders
customer_id = C_002        2 million orders
...

All rows for C_HOT generally land in one hash partition. One task now processes hundreds of millions of rows while other tasks finish quickly.

This is why adding more executors does not necessarily solve skew. The skewed key still belongs to one partition unless the execution strategy changes.

A Runnable PySpark Example

The following example creates a skewed orders dataset:

One customer, C_HOT, has 700 orders.
One hundred normal customers have 3 orders each.
The customer dimension is small enough to broadcast.
The expected joined row count is 1000.
The expected revenue is 13000.

Run it with a local Spark installation:

pip install pyspark

from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("partitioning-skew-broadcast-demo")
    .config("spark.sql.shuffle.partitions", "8")
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)

# 700 orders from one unusually active customer.
hot_orders = [
    (f"HOT-{i:04d}", "C_HOT", 10)
    for i in range(700)
]

# 100 normal customers, each with 3 orders.
normal_orders = [
    (f"NORMAL-{customer_num:03d}-{order_num}", f"C_{customer_num:03d}", 20)
    for customer_num in range(1, 101)
    for order_num in range(3)
]

orders = spark.createDataFrame(
    hot_orders + normal_orders,
    ["order_id", "customer_id", "amount"]
)

customers = spark.createDataFrame(
    [("C_HOT", "enterprise")]
    + [(f"C_{i:03d}", "standard") for i in range(1, 101)],
    ["customer_id", "segment"]
)

# Force a hash shuffle by customer_id so we can observe skew.
orders_by_customer = orders.repartition(8, "customer_id").cache()
orders_by_customer.count()  # Materialize the cached shuffled DataFrame.

partition_sizes = (
    orders_by_customer.rdd
    .mapPartitions(lambda rows: [sum(1 for _ in rows)])
    .collect()
)

print("Rows per partition:", partition_sizes)
print("Largest partition:", max(partition_sizes))

# Broadcast the small dimension table.
joined = (
    orders_by_customer
    .join(F.broadcast(customers), on="customer_id", how="inner")
)

result = (
    joined
    .agg(
        F.count("*").alias("joined_rows"),
        F.sum("amount").alias("total_revenue")
    )
    .first()
)

print("Joined rows:", result["joined_rows"])
print("Total revenue:", result["total_revenue"])

assert result["joined_rows"] == 1000
assert result["total_revenue"] == 13000
assert max(partition_sizes) >= 700

joined.explain(mode="formatted")

spark.stop()

The exact list of partition sizes varies because Spark’s hash partitioning determines where each customer key lands. However, one partition must contain at least the 700 C_HOT rows.

The important result is that the join uses a broadcast strategy for customers. In the physical plan, look for an operator similar to:

BroadcastHashJoin

Spark sends the small customers table to each executor. The large orders DataFrame does not need to be shuffled just to perform this join.

When Broadcast Joins Help—and When They Do Not

A broadcast join is appropriate when one join input is genuinely small after filtering and selecting only needed columns.

orders.join(F.broadcast(customers), "customer_id")

Conceptually, Spark does this:

Large orders partitions        Small customers table
        |                              |
        |                         Broadcast to all executors
        v                              v
Local join in each executor partition

The main benefit is avoiding a shuffle of the large table for the join.

However, broadcast joins do not solve every skew problem:

A skewed scan remains skewed.

If one input file is much larger than the rest, one scan task may still lag.
A later aggregation can still skew.

Joining orders to customers may be fast, but this can still produce a skewed shuffle:

   joined.groupBy("customer_id").sum("amount")

All C_HOT rows must still be brought together for that aggregation.

Broadcasting a table that is too large can fail. Every executor needs memory for the broadcast relation. A dimension that looks small in compressed Parquet can expand significantly after deserialization and hashing.

Spark can automatically broadcast relations below spark.sql.autoBroadcastJoinThreshold, but the threshold is configuration-dependent. Explicit broadcast() is useful when you understand the size and want to communicate intent, not as a blind optimization.

If you want a broader learning path after practicing these concepts, this optional data engineering roadmap for beginners can help place Spark optimization topics in context.

Detecting and Fixing Data Skew

Start by confirming that skew exists. In the Spark UI, look for a stage where:

Most tasks finish quickly.
One or a few tasks run much longer.
A few tasks read or write substantially more shuffle data.
Executor memory pressure or spill occurs only in a few tasks.

You can also inspect key frequency directly:

orders.groupBy("customer_id") \
    .count() \
    .orderBy(F.desc("count")) \
    .show(20, truncate=False)

Let Adaptive Query Execution Help First

Adaptive Query Execution (AQE) allows Spark to make some decisions after it observes runtime shuffle statistics.

Useful settings include:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

For eligible shuffle joins, Spark can split unusually large shuffle partitions into smaller pieces. AQE can also coalesce tiny shuffle partitions.

AQE is often the first mitigation to try because it requires no business-logic changes. But it cannot always split work for a single extremely hot key, especially when all records for that key must be processed together.

Use Key Salting for Severe Hot Keys

Salting splits one hot key into multiple artificial keys.

For example, instead of joining only on:

C_HOT

you create:

C_HOT_0
C_HOT_1
...
C_HOT_7

The fact table assigns a deterministic salt:

salt_count = 8

salted_orders = orders.withColumn(
    "salt",
    F.when(
        F.col("customer_id") == "C_HOT",
        F.pmod(F.hash("order_id"), F.lit(salt_count))
    ).otherwise(F.lit(0))
)

The dimension table must duplicate only the hot key across the same salt values. Normal customer rows keep salt 0. Then join on both customer_id and salt.

Salting is most useful when both join inputs are too large to broadcast. It adds complexity and duplicates dimension rows, so do not apply it to every key. Identify the small set of problematic keys first.

Production Pitfalls

A few common choices create avoidable Spark performance problems.

Using repartition() everywhere.

Every repartition is a shuffle. Repartition only when it improves a downstream operation, such as balancing data before a large write or aligning work before a non-broadcast join.

Using too many shuffle partitions.

Thousands of tiny tasks add scheduler overhead and can produce small output files. Too few partitions create large tasks, memory pressure, and poor parallelism. Tune based on observed input size, shuffle volume, task duration, and cluster capacity.

Assuming a broadcast hint is always safe.

A forced broadcast can cause executor out-of-memory failures. Filter dimensions early, select only necessary columns, and verify size under realistic production data volumes.

Writing files partitioned by high-cardinality keys.

Writing data partitioned by customer_id can create massive directory and small-file counts. Storage partitions should usually reflect common pruning filters, such as date, region, or source system.

Ignoring post-join cardinality.

A duplicate key in a dimension table can turn a one-to-many join into a many-to-many join. This both inflates results and amplifies skew.

Testing Partitioning and Join Behavior

Test correctness before tuning performance.

For joins, create assertions for:

Expected row counts
Expected aggregate totals
Null handling for unmatched keys
Duplicate dimension keys
Known hot-key behavior

For example, the runnable example asserts both row count and total revenue. In a real pipeline, add a test where a dimension intentionally contains duplicate customer_id values and verify that the pipeline rejects or deduplicates them according to your business rule.

For performance regression testing:

Build a representative sample containing hot keys.
Run the job with Spark event logging enabled.
Compare shuffle read/write volume, task duration distribution, spill metrics, and output row counts.
Validate the physical plan with explain("formatted").
Test with production-like executor memory and partition counts; local mode cannot reveal all distributed failure modes.

Takeaway

Spark partitioning determines how work is distributed, while joins and aggregations often force data to shuffle. A skewed key can make one partition dominate an entire stage.

Use broadcast joins when one side is safely small, because they avoid shuffling the large side. Use AQE to handle many skewed shuffle joins automatically. For persistent, severe hot keys that cannot be broadcast, use targeted salting and validate both correctness and resource usage.

The best optimization is not “more partitions” by default—it is choosing a partitioning and join strategy that matches the actual shape of your data.

Designing Airflow DAGs with Retries, SLAs, and Idempotency

pawan deore — Thu, 05 Nov 2026 13:54:44 +0000

Designing Airflow DAGs with Retries, SLAs, and Idempotency

A production Airflow DAG must handle ordinary failure without corrupting data or silently missing its delivery target. Three design concerns work together:

Retries handle transient failures such as temporary database locks, API rate limits, or network interruptions.
SLAs make late completion visible to operators.
Idempotency ensures that rerunning or retrying a task produces the same correct result instead of duplicates.

This tutorial uses a daily revenue aggregation pipeline as an example. It assumes:

Apache Airflow 2.x, where classic task-level SLA monitoring is available.
PostgreSQL as the warehouse.
A source table named orders and a reporting table named daily_order_revenue.
A daily DAG that processes a closed data interval.

Airflow 3 introduces newer alerting concepts such as Deadline Alerts, so verify the behavior of SLA-related settings against the version you operate.

The architecture: separate orchestration from data correctness

Consider this pipeline:

PostgreSQL orders table
        |
        |  Read one Airflow data interval
        v
daily_revenue DAG task
        |
        |  Delete and rebuild one report partition
        v
PostgreSQL daily_order_revenue table
        |
        +--> Retry policy handles temporary failures
        +--> SLA callback alerts when completion is late

The task processes one interval at a time:

[2025-03-10 00:00:00 UTC, 2025-03-11 00:00:00 UTC)

That interval is more important than the wall-clock time at which the task happens to run. A manual rerun next week should still process the same logical interval, not “today.”

This leads to a useful division of responsibility:

Concern	Responsibility
Which interval should be processed?	Airflow schedule and data interval
How should transient failure be handled?	Retry configuration
What happens if the same interval runs twice?	Idempotent warehouse write
How do operators learn that output is late?	SLA monitoring or deadline alerting
How is permanent bad input handled?	Validation and explicit task failure

Retries are not a substitute for idempotency. A retry means the task may execute its write logic multiple times. If each execution appends rows without a deterministic key or replacement strategy, the retry itself can create incorrect output.

Start your data engineering journey with Data Engineering: A Complete Roadmap for Beginners cuts through the noise and explains the essentials in a friendly, beginner-focused way across 15 comprehensive chapters and 190 pages.

Model retries around failure types

Not every error deserves a retry.

A retry is useful when the next attempt may reasonably succeed without changing the code or input:

Temporary network failure to the database
Database deadlock or lock timeout
A short-lived warehouse outage
Rate limiting from an external API
Object storage request timeout

A retry is usually wasteful for deterministic failures:

Missing required source column
Invalid SQL syntax
Bad credentials
A malformed input file
A violated business rule, such as negative revenue when negatives are forbidden

Airflow retries are configured per task. In this example:

retries=3
retry_delay=timedelta(minutes=5)
retry_exponential_backoff=True
max_retry_delay=timedelta(minutes=30)

This means Airflow makes the initial attempt plus up to three retry attempts. Exponential backoff avoids having many failed tasks retry at the same moment after a shared outage.

Also use execution_timeout. A retry policy cannot help a task that hangs forever waiting on a database operation.

A good starting point is:

Keep retry counts modest.
Use a short initial delay for likely transient failures.
Cap exponential backoff.
Set an execution timeout based on the expected task duration.
Make writes safe to repeat before enabling retries.

Build an idempotent daily aggregation task

The following DAG task rebuilds a single reporting partition. It uses Airflow’s data_interval_start and data_interval_end instead of datetime.now().

The destination table has a primary key on (report_date, order_id), but the task also deletes the entire daily partition before inserting. That deletion handles a subtle case: an order that was present in a prior attempt but is no longer eligible in a later rerun.

Example schema

CREATE TABLE orders (
    order_id BIGINT PRIMARY KEY,
    paid_at TIMESTAMPTZ NOT NULL,
    amount_cents INTEGER NOT NULL,
    status TEXT NOT NULL
);

CREATE TABLE daily_order_revenue (
    report_date DATE NOT NULL,
    order_id BIGINT NOT NULL,
    amount_cents INTEGER NOT NULL,
    loaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (report_date, order_id)
);

Airflow DAG

Save this as dags/daily_revenue.py. It requires a PostgreSQL Airflow connection named warehouse_postgres.

from datetime import datetime, timedelta, timezone
import logging

from airflow.decorators import dag, task
from airflow.operators.python import get_current_context
from airflow.providers.postgres.hooks.postgres import PostgresHook


def report_sla_miss(dag, task_list, blocking_task_list, slas, blocking_tis):
    logging.error(
        "SLA missed for DAG=%s tasks=%s blocking_tasks=%s",
        dag.dag_id,
        task_list,
        blocking_task_list,
    )


@dag(
    dag_id="daily_revenue",
    start_date=datetime(2025, 3, 1, tzinfo=timezone.utc),
    schedule="0 0 * * *",
    catchup=False,
    max_active_runs=1,
    sla_miss_callback=report_sla_miss,
    tags=["reporting", "postgres"],
)
def daily_revenue():

    @task(
        retries=3,
        retry_delay=timedelta(minutes=5),
        retry_exponential_backoff=True,
        max_retry_delay=timedelta(minutes=30),
        execution_timeout=timedelta(minutes=20),

        # See the SLA section for why this is 26 hours.
        sla=timedelta(hours=26),
    )
    def rebuild_daily_revenue():
        context = get_current_context()

        interval_start = context["data_interval_start"]
        interval_end = context["data_interval_end"]
        report_date = interval_start.date()

        sql = """
        DELETE FROM daily_order_revenue
        WHERE report_date = %s;

        INSERT INTO daily_order_revenue (
            report_date,
            order_id,
            amount_cents
        )
        SELECT
            %s AS report_date,
            order_id,
            amount_cents
        FROM orders
        WHERE paid_at >= %s
          AND paid_at < %s
          AND status = 'paid';
        """

        hook = PostgresHook(postgres_conn_id="warehouse_postgres")

        # Use one database transaction: either replacement succeeds fully,
        # or the previous partition remains unchanged after rollback.
        with hook.get_conn() as conn:
            with conn.cursor() as cursor:
                cursor.execute(
                    sql,
                    (
                        report_date,
                        report_date,
                        interval_start,
                        interval_end,
                    ),
                )
            conn.commit()

    rebuild_daily_revenue()


daily_revenue()

Suppose the DAG run processes 2025-03-10 and the source contains:

order_id	paid_at	amount_cents	status
101	2025-03-10 08:15:00+00	2599	paid
102	2025-03-10 11:40:00+00	4999	paid
103	2025-03-10 15:00:00+00	1200	cancelled
104	2025-03-11 00:00:00+00	3000	paid

The expected destination rows are:

report_date	order_id	amount_cents
2025-03-10	101	2599
2025-03-10	102	4999

Order 103 is excluded because it was cancelled. Order 104 is excluded because the interval is half-open: it starts exactly at the next day’s boundary.

If the task fails after the DELETE but before the INSERT, the database transaction rolls back. If Airflow retries after a successful insert, the retry deletes and recreates the same partition. Both outcomes are safe.

Understand SLA timing before setting an SLA

An SLA is an operational expectation, not a data-quality guarantee. In classic Airflow SLA behavior, a missed SLA can trigger a callback or notification, but it does not automatically stop a task, repair data, or create a retry.

The confusing part is the reference time.

For a daily schedule at midnight, a run that executes shortly after:

2025-03-11 00:00 UTC

typically processes this interval:

2025-03-10 00:00 UTC through 2025-03-11 00:00 UTC

Its logical date is associated with the start of the interval: 2025-03-10 00:00 UTC.

If the business requirement is:

“Yesterday’s revenue must be ready by 02:00 UTC on the next calendar day.”

Then the deadline is 26 hours after the interval start, not 2 hours after it. That is why the example uses:

sla=timedelta(hours=26)

An sla=timedelta(hours=2) setting would be interpreted relative to the logical date and may already be overdue when the scheduled run begins.

Classic SLA monitoring also depends on a healthy scheduler. Treat SLA alerts as a signal to investigate, not as the only mechanism for delivery monitoring.

For a broader learning path that connects Airflow concepts with warehouse modeling and orchestration fundamentals, this optional data engineering roadmap can be a useful next step.

Avoid the “idempotent SQL but changing source” trap

The partition replacement pattern is idempotent with respect to a fixed source state. But source state can change.

For example:

The first attempt reads an order with status = 'paid'.
The task fails after committing the destination write, or is manually rerun later.
Someone changes the source order to status = 'refunded'.
The rerun produces a different daily report.

That may be correct or incorrect depending on the business definition.

Decide explicitly whether your report is:

As-of processing time: reruns should use the latest source state.
As-of interval close: reruns should reproduce the original result.
Eventually corrected: late arrivals and corrections are expected, and recent partitions are intentionally rebuilt.

For reproducible historical output, use immutable source files, source snapshots, change-data-capture offsets, or a versioned bronze layer. Airflow retries alone cannot freeze mutable source data.

Also avoid using NOW() inside transformation predicates. This query is unsafe for backfills:

WHERE paid_at >= NOW() - INTERVAL '1 day'

It produces a different answer each time it runs. Use the Airflow interval boundaries instead.

Test retries, writes, and schedules separately

Test the transformation SQL independently from Airflow. Insert representative rows around boundaries:

Exactly at data_interval_start
One microsecond before data_interval_end
Exactly at data_interval_end
Cancelled or refunded orders
Duplicate source events, if your source allows them
Late-arriving events

Then verify these properties:

One interval produces only expected rows.
Running the same interval twice produces the same destination state.
A failure between delete and insert leaves no partial partition.
Adjacent intervals do not overlap or leave gaps.
A backfill uses historical interval boundaries, not the current date.

For the transactional behavior, deliberately raise an exception after the DELETE in a non-production test environment. Confirm that the transaction rolls back and the prior partition remains available.

For DAG-level tests, Airflow can parse and validate the DAG without executing it:

airflow dags list | grep daily_revenue
airflow dags test daily_revenue 2025-03-11

The dags test command is useful for validating task execution for a logical date, but it is not a replacement for an integration test database. Use an isolated PostgreSQL schema or disposable database for tests that mutate tables.

Finally, test SLA callbacks differently from task logic. SLA behavior is scheduler-driven and time-dependent, so unit test the callback function directly and validate actual alert routing in a staging Airflow environment.

Production pitfalls to plan for

A few configuration choices often cause surprising behavior:

catchup=False does not make backfills impossible. Manual runs can still be created, so interval-based SQL remains necessary.
max_active_runs=1 protects a mutable daily partition table. Without it, overlapping runs for the same business date can race unless the write strategy is designed for concurrency.
Retries can overload a failing dependency. Use exponential backoff and consider Airflow pools for shared databases or APIs.
Do not put secrets in DAG code. Use Airflow connections, a secrets backend, or workload identity.
A task can succeed with bad data. Add validation tasks for row counts, null rates, reconciliation totals, or expected freshness.
SLA misses do not imply data loss. A task may finish late but correctly; conversely, a task can finish on time with incorrect results.

Takeaway

Reliable Airflow DAGs treat retries, SLAs, and idempotency as one design problem:

Retry only failures likely to be transient.
Use data intervals, not current timestamps, to define work.
Make destination writes safe to repeat with transactions and deterministic partition replacement or upserts.
Set SLAs relative to Airflow’s logical date semantics.
Test reruns, interval boundaries, rollback behavior, and mutable-source assumptions.

When a task can be retried or manually rerun without changing the correctness of its output, Airflow becomes much safer to operate under real-world failure conditions.

CDC with Debezium and Kafka: A Source-to-Sink Walkthrough

pawan deore — Tue, 03 Nov 2026 02:18:45 +0000

CDC with Debezium and Kafka: A Source-to-Sink Walkthrough

Change Data Capture (CDC) moves database changes as events instead of repeatedly querying whole tables. In this walkthrough, PostgreSQL is the source of truth, Debezium reads its write-ahead log (WAL), Kafka transports the change events, and a small Python consumer maintains a SQLite projection as the sink.

The example uses a customers table, but the same pattern applies to operational databases feeding search indexes, warehouses, caches, downstream services, or lakehouse ingestion pipelines.

The Architecture and Delivery Model

The pipeline has four responsibilities:

PostgreSQL
   │
   │ logical decoding from WAL
   ▼
Debezium PostgreSQL connector
   │
   │ Kafka Connect source task
   ▼
Kafka topic: shop.public.customers
   │
   │ ordered per Kafka partition
   ▼
Python sink consumer
   │
   ▼
SQLite customer_projection

PostgreSQL records every committed change in its WAL. Debezium does not poll SELECT * FROM customers; instead, it creates or uses a logical replication slot and converts committed WAL changes into Kafka records.

A typical lifecycle is:

Debezium performs an initial snapshot of the configured table.
It records the WAL position associated with that snapshot.
It streams later inserts, updates, and deletes from the WAL.
A sink consumes Kafka records and updates its own representation of the data.

This architecture is usually at-least-once end to end. A consumer can process the same event again after a restart, rebalance, or failure between writing to the sink and committing its Kafka offset. Therefore, sink writes must be idempotent.

For a table-backed sink, the usual idempotent pattern is:

use the source primary key as the Kafka message key;
upsert inserts and updates;
delete by primary key;
commit the Kafka offset only after the sink transaction commits.

Prepare PostgreSQL for Logical Replication

Assume the following:

PostgreSQL is reachable at localhost:5432;
Kafka is reachable at localhost:9092;
Kafka Connect is running at http://localhost:8083;
the Kafka Connect worker has the Debezium PostgreSQL connector plugin installed;
PostgreSQL has wal_level=logical and has been restarted after that setting was changed.

Run the following as a PostgreSQL administrator. The ALTER SYSTEM command requires a database restart before it takes effect.

ALTER SYSTEM SET wal_level = 'logical';

CREATE ROLE debezium
  WITH LOGIN REPLICATION PASSWORD 'dbz_password';

CREATE DATABASE appdb;

Connect to appdb, then create a source table and initial rows:

CREATE TABLE public.customers (
    id         INTEGER PRIMARY KEY,
    full_name  TEXT NOT NULL,
    email      TEXT NOT NULL UNIQUE,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

INSERT INTO public.customers (id, full_name, email)
VALUES
    (1, 'Aisha Rahman', 'aisha@example.com'),
    (2, 'Mark Chen', 'mark@example.com');

GRANT USAGE ON SCHEMA public TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;

CREATE PUBLICATION dbz_publication
FOR TABLE public.customers;

The publication tells PostgreSQL which table changes may be emitted through the pgoutput logical decoding plugin.

The primary key is important. Debezium uses it to construct a stable Kafka key such as:

{"id": 1}

Without a stable key, a downstream sink cannot reliably identify which row should be updated or deleted.

Register the Debezium Connector

Kafka Connect runs connectors. Debezium is a source connector: it reads from PostgreSQL and writes to Kafka.

Save the following as customers-connector.json:

{
  "name": "customers-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",

    "database.hostname": "localhost",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "dbz_password",
    "database.dbname": "appdb",

    "topic.prefix": "shop",
    "table.include.list": "public.customers",

    "publication.name": "dbz_publication",
    "publication.autocreate.mode": "disabled",
    "slot.name": "customers_cdc_slot",

    "snapshot.mode": "initial",

    "schema.history.internal.kafka.bootstrap.servers": "localhost:9092",
    "schema.history.internal.kafka.topic": "dbz-schema-history.shop",

    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",

    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.add.fields": "op,source.ts_ms",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "transforms.unwrap.delete.tombstone.handling.mode": "drop"
  }
}

curl -X POST \
  -H "Content-Type: application/json" \
  --data @customers-connector.json \
  http://localhost:8083/connectors

Check connector and task status:

curl http://localhost:8083/connectors/customers-cdc/status

The source topic is:

shop.public.customers

Without the ExtractNewRecordState transformation, Debezium emits a detailed envelope containing before, after, source metadata, transaction context, and operation type. That full envelope is valuable for audit pipelines.

This tutorial unwraps the after state because the sink is maintaining a current-state table. It adds metadata fields such as __op, where values commonly include:

r: snapshot read
c: create
u: update
d: delete

For a broader structured learning path after building this example, this data engineering roadmap is an optional next step.

Generate Changes and Inspect the Kafka Records

Make a few source-side changes:

UPDATE public.customers
SET email = 'aisha.rahman@example.com',
    updated_at = now()
WHERE id = 1;

INSERT INTO public.customers (id, full_name, email)
VALUES (3, 'Omar Haddad', 'omar@example.com');

DELETE FROM public.customers
WHERE id = 2;

After unwrapping, the update event is approximately:

{
  "id": 1,
  "full_name": "Aisha Rahman",
  "email": "aisha.rahman@example.com",
  "updated_at": "2025-...",
  "__op": "u",
  "__source_ts_ms": 1730000000000
}

The exact timestamps will differ. A delete is rewritten into a value containing a deletion marker, typically __deleted: "true", while retaining the Kafka key.

You can inspect records using Kafka CLI tools. The exact command depends on the Kafka distribution, but the essential requirement is to display both key and value:

kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic shop.public.customers \
  --from-beginning \
  --property print.key=true \
  --property key.separator=" | "

Build a Small Idempotent Python Sink

This consumer writes the topic into SQLite. It is intentionally small, but it demonstrates the core production rule: write to the sink first, then commit the Kafka offset.

Install the dependency:

pip install kafka-python

Save this as customer_sink.py:

import json
import sqlite3
from kafka import KafkaConsumer

TOPIC = "shop.public.customers"

db = sqlite3.connect("customer_projection.db")
db.execute("""
CREATE TABLE IF NOT EXISTS customer_projection (
    id INTEGER PRIMARY KEY,
    full_name TEXT NOT NULL,
    email TEXT NOT NULL,
    updated_at TEXT NOT NULL
)
""")
db.commit()

consumer = KafkaConsumer(
    TOPIC,
    bootstrap_servers="localhost:9092",
    group_id="sqlite-customer-sink-v1",
    auto_offset_reset="earliest",
    enable_auto_commit=False,
    key_deserializer=lambda raw: json.loads(raw.decode("utf-8")) if raw else None,
    value_deserializer=lambda raw: json.loads(raw.decode("utf-8")) if raw else None,
)

for message in consumer:
    key = message.key
    value = message.value

    if not key or "id" not in key:
        raise ValueError(f"Missing primary-key message key: {key}")

    customer_id = key["id"]
    is_delete = (
        value is None
        or str(value.get("__deleted", "")).lower() == "true"
    )

    with db:
        if is_delete:
            db.execute(
                "DELETE FROM customer_projection WHERE id = ?",
                (customer_id,)
            )
        else:
            db.execute("""
                INSERT INTO customer_projection (id, full_name, email, updated_at)
                VALUES (?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    full_name = excluded.full_name,
                    email = excluded.email,
                    updated_at = excluded.updated_at
            """, (
                customer_id,
                value["full_name"],
                value["email"],
                value["updated_at"],
            ))

    # Offset is committed only after SQLite commits successfully.
    consumer.commit()

Run it:

python customer_sink.py

If the consumer reads the initial snapshot followed by the three changes above, query the sink database:

sqlite3 customer_projection.db \
  "SELECT id, full_name, email FROM customer_projection ORDER BY id;"

Expected result:

1|Aisha Rahman|aisha.rahman@example.com
3|Omar Haddad|omar@example.com

id=2 is absent because its source row was deleted.

The upsert makes repeated create, snapshot, and update events safe. The delete is also idempotent: deleting an already absent row is harmless.

Handle Ordering, Duplicates, and Schema Changes Deliberately

Kafka guarantees order only within one partition. Records with the same non-null key are normally routed to the same partition, so all changes for customer id=1 retain order.

Do not assume a global order across customers. An update to customer 1 and an insert for customer 3 may be processed in different partitions and in a different relative order than another system observes them.

Duplicates are normal under at-least-once delivery. The SQLite sample handles them because:

inserts are upserts;
updates overwrite the same primary key;
deletes are repeatable;
the consumer commits offsets after the sink write.

Schema evolution needs an explicit plan. Adding a nullable source column is usually straightforward, but dropping or renaming columns can break consumers. For production pipelines:

version event contracts;
deploy sink support before emitting a required new field;
avoid SELECT *-style assumptions in consumers;
use a schema registry and compatibility rules if multiple services consume the topic.

The JSON converter simplifies this tutorial, but it does not provide strong schema governance. Avro, Protobuf, or JSON Schema with a registry is often preferable for shared event streams.

Production Pitfalls and Operational Controls

A replication slot retains WAL until Debezium confirms it has consumed it. If Kafka Connect is down for too long, PostgreSQL disk usage can grow because old WAL segments cannot be removed. Monitor replication-slot lag and WAL disk usage.

Other common pitfalls include:

No primary key: deletes and updates become difficult to apply safely downstream.
Snapshot surprises: an initial snapshot can produce many records. Ensure the sink can handle the initial volume and that topic retention is sufficient.
Connector restarts: source offsets are stored by Kafka Connect. Losing Connect internal topics can cause unwanted re-snapshots or duplicate processing.
Poison messages: one malformed record can repeatedly fail a consumer. Production consumers should validate records, log context, and use a dead-letter strategy.
Delete semantics: decide whether sinks need hard deletes, soft deletes, tombstones, or historical retention.
Database permissions: grant Debezium only the replication and read permissions it needs.
Topic management: pre-create production topics with intentional partition counts, retention, compaction choices, ACLs, and monitoring.

For a current-state projection, consider log compaction on the CDC topic only after understanding delete tombstones and consumer replay requirements. A compacted topic is useful for rebuilding latest state, while a retained append-only topic is more useful for historical replay.

Test the Pipeline and Verify Recovery Behavior

Test CDC as a data contract, not just a connectivity check.

Start with an integration test environment containing PostgreSQL, Kafka, Kafka Connect, and a disposable sink. Then verify:

Initial snapshot: source rows appear in the sink.
Insert: a new source row appears once logically in the sink.
Update: changed values replace old values.
Delete: the sink removes or marks the row according to your contract.
Restart recovery: stop the sink after a database write but before an offset commit, then restart it. The duplicate event must not corrupt the result.
Connector recovery: restart Kafka Connect and confirm streaming resumes from stored offsets.
Schema change: add a nullable column and verify consumers tolerate it.

A useful assertion compares source and sink after all events have settled:

SELECT id, full_name, email
FROM public.customers
ORDER BY id;

Compare that result with the sink projection query. For this current-state sink, they should match on the columns intentionally replicated.

Takeaway

Debezium CDC turns committed PostgreSQL changes into Kafka events by reading the WAL, not by polling tables. A reliable source-to-sink pipeline depends on stable primary keys, idempotent sink writes, offset commits after durable writes, and monitoring of replication-slot lag.

The smallest useful mental model is: PostgreSQL commits a change, Debezium publishes it, Kafka retains it, and the sink repeatedly applies it safely until its state converges with the source.

Build an Incremental ETL Pipeline in Python Using PostgreSQL Watermarks

pawan deore — Mon, 02 Nov 2026 17:20:50 +0000

Build an Incremental ETL Pipeline in Python Using PostgreSQL Watermarks

Incremental ETL avoids rereading an entire source table on every run. Instead, the pipeline remembers the latest source change it successfully loaded, then requests only records after that point. That remembered position is the watermark.

This tutorial builds a small but production-oriented pipeline that copies app.orders into warehouse.fact_orders. It uses PostgreSQL timestamps plus a primary-key tie-breaker as the watermark and performs idempotent upserts in Python.

Define the Watermark Contract First

A watermark is only reliable when its source column has a clear contract. In this example, each source row has:

updated_at: when the application last changed the row
id: a stable, unique primary key

The checkpoint is the pair:

(last_updated_at, last_id)

Rows are ordered lexicographically:

(updated_at, id)

For example, if the saved watermark is:

(2025-03-08 10:00:00+00, 42)

the next extraction includes rows where:

(updated_at, id) > ('2025-03-08 10:00:00+00', 42)

The id is important. Multiple rows can receive the same timestamp, especially when inserts happen in a batch. Without a tie-breaker, rows sharing the previous maximum timestamp may be skipped.

This design assumes:

updated_at is populated by the database or application for every insert and update.
updated_at is never intentionally moved backward.
Source rows are not physically deleted without another way to represent deletion.
The pipeline can safely reload a small overlap window.

The final assumption matters because timestamp watermarks are not a full change data capture system. We will revisit that limitation later.

Create Source, Target, and Pipeline-State Tables

For a small local example, use one PostgreSQL database with separate schemas. In a real system, the source and warehouse may be different databases, clusters, or services.

Run this SQL first:

CREATE SCHEMA IF NOT EXISTS app;
CREATE SCHEMA IF NOT EXISTS warehouse;

CREATE TABLE IF NOT EXISTS app.orders (
    id BIGSERIAL PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    status TEXT NOT NULL,
    total_cents INTEGER NOT NULL CHECK (total_cents >= 0),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

-- Ensure application updates refresh the watermark column.
CREATE OR REPLACE FUNCTION app.set_updated_at()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at := clock_timestamp();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS orders_set_updated_at ON app.orders;

CREATE TRIGGER orders_set_updated_at
BEFORE UPDATE ON app.orders
FOR EACH ROW
EXECUTE FUNCTION app.set_updated_at();

CREATE TABLE IF NOT EXISTS warehouse.fact_orders (
    order_id BIGINT PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    status TEXT NOT NULL,
    total_cents INTEGER NOT NULL,
    source_updated_at TIMESTAMPTZ NOT NULL,
    loaded_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

CREATE TABLE IF NOT EXISTS warehouse.pipeline_state (
    pipeline_name TEXT PRIMARY KEY,
    last_updated_at TIMESTAMPTZ NOT NULL,
    last_id BIGINT NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

INSERT INTO app.orders (customer_id, status, total_cents)
VALUES
    (101, 'paid', 2599),
    (102, 'pending', 4999),
    (103, 'paid', 1200);

The target table retains source_updated_at. That field is useful both for observability and for preventing an older reprocessed record from overwriting a newer target version.

Understand the Pipeline Architecture and Transaction Boundaries

The pipeline has four logical components:

PostgreSQL source table
        |
        |  rows after saved watermark
        v
Python extraction and transformation
        |
        |  idempotent upsert
        v
PostgreSQL warehouse table
        |
        |  successful load only
        v
PostgreSQL pipeline_state checkpoint

The critical correctness rule is:

Advance the watermark only after the target rows have been written successfully.

If the process fails during loading, the saved checkpoint must not move. On the next execution, the same source rows are read again. That requires target writes to be idempotent: loading the same source row twice must produce the same final target state.

This example uses a transaction around target upserts and the state update. It also obtains a PostgreSQL advisory lock so two scheduler invocations cannot run the same pipeline concurrently.

The source extraction uses a REPEATABLE READ transaction. That gives the run a consistent source snapshot: the upper watermark and extracted rows describe the same view of the source table.

If you want a broader structured learning path after implementing this pattern, this data engineering roadmap is an optional next step.

Run the Incremental Python ETL Job

Install the PostgreSQL driver:

pip install "psycopg[binary]"

Set a connection string. This example uses the same database for source and target:

export DATABASE_URL="postgresql://postgres:postgres@localhost:5432/analytics"

Save the following as incremental_orders.py:

import os
from datetime import datetime, timedelta, timezone

import psycopg
from psycopg.rows import dict_row

DATABASE_URL = os.environ["DATABASE_URL"]
PIPELINE_NAME = "orders_incremental_v1"

# Re-read recent changes to tolerate slightly late commits.
# Target upserts make this safe.
LOOKBACK = timedelta(minutes=10)

INITIAL_WATERMARK = datetime(1970, 1, 1, tzinfo=timezone.utc)


def run_pipeline():
    with (
        psycopg.connect(DATABASE_URL, row_factory=dict_row) as source_conn,
        psycopg.connect(DATABASE_URL, row_factory=dict_row) as target_conn,
    ):
        # This transaction contains both target writes and state advancement.
        with target_conn.transaction():
            with target_conn.cursor() as target_cur:
                # Prevent concurrent runs of this exact pipeline.
                target_cur.execute(
                    "SELECT pg_advisory_xact_lock(hashtext(%s))",
                    (PIPELINE_NAME,),
                )

                target_cur.execute(
                    """
                    SELECT last_updated_at, last_id
                    FROM warehouse.pipeline_state
                    WHERE pipeline_name = %s
                    FOR UPDATE
                    """,
                    (PIPELINE_NAME,),
                )
                saved_state = target_cur.fetchone()

                if saved_state:
                    # Intentionally overlap prior work. id=0 means all IDs at
                    # the overlap timestamp range are eligible for reload.
                    lower_updated_at = saved_state["last_updated_at"] - LOOKBACK
                    lower_id = 0
                else:
                    lower_updated_at = INITIAL_WATERMARK
                    lower_id = 0

                # Read a consistent source snapshot.
                with source_conn.transaction():
                    with source_conn.cursor() as source_cur:
                        source_cur.execute(
                            "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY"
                        )

                        # Capture an upper bound from the same source snapshot.
                        source_cur.execute(
                            """
                            SELECT updated_at, id
                            FROM app.orders
                            ORDER BY updated_at DESC, id DESC
                            LIMIT 1
                            """
                        )
                        upper_bound = source_cur.fetchone()

                        if upper_bound is None:
                            print("Source table is empty; nothing to load.")
                            return

                        upper_updated_at = upper_bound["updated_at"]
                        upper_id = upper_bound["id"]

                        source_cur.execute(
                            """
                            SELECT
                                id,
                                customer_id,
                                status,
                                total_cents,
                                updated_at AS source_updated_at
                            FROM app.orders
                            WHERE (updated_at, id) > (%s, %s)
                              AND (updated_at, id) <= (%s, %s)
                            ORDER BY updated_at, id
                            """,
                            (
                                lower_updated_at,
                                lower_id,
                                upper_updated_at,
                                upper_id,
                            ),
                        )
                        rows = source_cur.fetchall()

                # Upsert makes retries and overlap reads safe.
                if rows:
                    target_cur.executemany(
                        """
                        INSERT INTO warehouse.fact_orders (
                            order_id,
                            customer_id,
                            status,
                            total_cents,
                            source_updated_at
                        )
                        VALUES (
                            %(id)s,
                            %(customer_id)s,
                            %(status)s,
                            %(total_cents)s,
                            %(source_updated_at)s
                        )
                        ON CONFLICT (order_id) DO UPDATE
                        SET
                            customer_id = EXCLUDED.customer_id,
                            status = EXCLUDED.status,
                            total_cents = EXCLUDED.total_cents,
                            source_updated_at = EXCLUDED.source_updated_at,
                            loaded_at = clock_timestamp()
                        WHERE warehouse.fact_orders.source_updated_at
                              <= EXCLUDED.source_updated_at
                        """,
                        rows,
                    )

                # Advance state only after all target upserts succeeded.
                target_cur.execute(
                    """
                    INSERT INTO warehouse.pipeline_state (
                        pipeline_name,
                        last_updated_at,
                        last_id
                    )
                    VALUES (%s, %s, %s)
                    ON CONFLICT (pipeline_name) DO UPDATE
                    SET
                        last_updated_at = EXCLUDED.last_updated_at,
                        last_id = EXCLUDED.last_id,
                        updated_at = clock_timestamp()
                    """,
                    (PIPELINE_NAME, upper_updated_at, upper_id),
                )

                print(
                    f"Loaded or reprocessed {len(rows)} rows. "
                    f"New watermark: ({upper_updated_at}, {upper_id})"
                )


if __name__ == "__main__":
    run_pipeline()

Run it:

python incremental_orders.py

Verify Expected Results and Incremental Behavior

After the first run, query the target:

SELECT order_id, customer_id, status, total_cents
FROM warehouse.fact_orders
ORDER BY order_id;

Expected result:

order_id	customer_id	status	total_cents
1	101	paid	2599
2	102	pending	4999
3	103	paid	1200

The state table should contain one checkpoint:

SELECT pipeline_name, last_updated_at, last_id
FROM warehouse.pipeline_state;

Now simulate an update and a new order:

UPDATE app.orders
SET status = 'paid'
WHERE id = 2;

INSERT INTO app.orders (customer_id, status, total_cents)
VALUES (104, 'pending', 7500);

Run the Python job again. The pipeline will:

reread a small 10-minute overlap,
update order 2 from pending to paid,
insert the new order,
advance the state to the newest (updated_at, id) pair.

Because the load uses INSERT ... ON CONFLICT DO UPDATE, overlap rows do not create duplicates.

Production Pitfalls That Watermarks Do Not Solve Automatically

Late commits can be missed without overlap. An application transaction may assign updated_at, remain open for several minutes, and commit after the ETL has already moved beyond that timestamp. The overlap window reduces this risk, but it is not a mathematical guarantee for arbitrarily long transactions.

Backdated timestamps break the contract. If an application sets updated_at from client-provided data or restores old timestamps during a migration, the row may fall behind the checkpoint and never be extracted.

Hard deletes are invisible. A query for changed rows cannot detect a row that no longer exists. Use soft deletes such as deleted_at, maintain a deletion tombstone table, or adopt change data capture (CDC).

Timestamp precision matters. PostgreSQL TIMESTAMPTZ supports microseconds, but a timestamp alone is still not a unique event identifier. The (updated_at, id) pair handles different rows with identical timestamps; it does not represent multiple distinct updates to the same row with the exact same timestamp.

Large batches need staging. executemany is appropriate for a small tutorial. For high volume, load into a staging table with PostgreSQL COPY, then perform a set-based INSERT ... SELECT ... ON CONFLICT.

For stronger guarantees across long-running transactions, deletes, and every row-level change, use PostgreSQL logical replication/CDC or an application outbox rather than relying solely on an updated_at watermark.

Test the Pipeline as a State Machine

Test more than whether rows appear in the target. The state transition is the important behavior.

Useful integration tests include:

Initial load: insert source rows, run the job, and assert target row count and state values.
Idempotent retry: run the job twice without source changes; assert no duplicate target rows.
Update propagation: update a source order, rerun, and assert the target reflects the new values.
Same-timestamp ordering: insert multiple rows with a controlled identical updated_at; confirm all are loaded because id breaks ties.
Failure before checkpoint: intentionally fail after extraction but before the state update, then rerun. Target data may be reprocessed, but no source change should be lost.
Concurrent execution: start two jobs and confirm the advisory lock serializes them.

Keep test data small and inspect warehouse.pipeline_state directly. A pipeline can look correct in the destination while silently holding an incorrect checkpoint that causes future data loss.

Takeaway

A PostgreSQL watermark pipeline is built on a simple rule: extract rows after a durable checkpoint, load them idempotently, and update that checkpoint only after the load succeeds. Use a composite (updated_at, id) watermark, a bounded source snapshot, transactional target writes, and an overlap window. It is a practical pattern for mutable tables, but for strict handling of deletes and long-running transactions, move to CDC or an outbox-based design.

Build an MCP Server with FastMCP and Connect It to Claude

pawan deore — Mon, 17 Aug 2026 09:03:45 +0000

Model Context Protocol (MCP) makes it possible for AI assistants to interact with external tools and data sources in a standardized way.

In this tutorial, we'll build a simple MCP server using FastMCP, expose it over HTTPS with LocalTunnel, and connect it to Claude as a custom connector.

Our example will be a simple note-taking server with two tools:

get_notes — retrieve saved notes
add_note — add a new note

By the end, Claude will be able to read from and write to our MCP server.

🎥 Watch the Tutorial

What We'll Build

Claude
   │
   │ MCP
   ▼
HTTPS Tunnel
   │
   ▼
FastMCP Server
   │
   ▼
Mock Database

We'll use an in-memory list as our database for simplicity. In a real application, you could replace it with PostgreSQL, SQLite, MongoDB, or another database.

Prerequisites

You'll need:

Python 3
VS Code
Node.js and npm
A Claude account that supports custom connectors

Create a new project and open it in VS Code.

1. Create a Virtual Environment

Start by creating a Python virtual environment:

python3 -m venv venv

Activate it on macOS/Linux:

source venv/bin/activate

2. Install FastMCP

Install FastMCP using pip:

pip install fastmcp

FastMCP makes it much easier to create MCP servers in Python without implementing the MCP protocol yourself.

3. Create the MCP Server

Create a file named main.py.

Start by importing FastMCP and creating an MCP server:

from fastmcp import FastMCP

mcp = FastMCP("Simple Note Taker")

For this tutorial, we'll use a simple Python list as our temporary database:

notes = [
    {
        "id": 0,
        "text": "Remember to deploy server"
    },
    {
        "id": 1,
        "text": "Eat Apple at 5 PM"
    }
]

We now have two initial notes.

4. Create the `get_notes` Tool

FastMCP allows us to turn a normal Python function into an MCP tool using the @mcp.tool decorator.

@mcp.tool
def get_notes() -> str:
    output = ""

    for note in notes:
        output += f"Note #{note['id']}: {note['text']}\n"

    return output

This function loops through our notes and returns them as a string.

5. Create the `add_note` Tool

Next, let's create a tool that allows Claude to add a new note.

@mcp.tool
def add_note(text: str) -> str:
    new_id = len(notes)

    notes.append({
        "id": new_id,
        "text": text
    })

    return f"Successfully saved note #{new_id}"

The function accepts the note text, generates a new ID, stores the note, and returns a success message.

Our MCP server now exposes two tools:

Tool	Description
`get_notes`	Returns all saved notes
`add_note`	Adds a new note

6. Configure the MCP Server

We want our MCP server to run over HTTP on port 8000.

Add this to the bottom of main.py:

if __name__ == "__main__":
    mcp.run(
        transport="http",
        host="0.0.0.0",
        port=8000
    )

The complete main.py looks like this:

from fastmcp import FastMCP

mcp = FastMCP("Simple Note Taker")

notes = [
    {
        "id": 0,
        "text": "Remember to deploy server"
    },
    {
        "id": 1,
        "text": "Eat Apple at 5 PM"
    }
]


@mcp.tool
def get_notes() -> str:
    output = ""

    for note in notes:
        output += f"Note #{note['id']}: {note['text']}\n"

    return output


@mcp.tool
def add_note(text: str) -> str:
    new_id = len(notes)

    notes.append({
        "id": new_id,
        "text": text
    })

    return f"Successfully saved note #{new_id}"


if __name__ == "__main__":
    mcp.run(
        transport="http",
        host="0.0.0.0",
        port=8000
    )

7. Run the MCP Server

Start the server:

python3 main.py

FastMCP should start listening on port 8000.

Our MCP server is now running locally.

However, Claude cannot directly access localhost:8000, so we need to expose it through an HTTPS URL.

8. Create an HTTPS Tunnel with LocalTunnel

For this tutorial, we'll use LocalTunnel.

Install it globally:

npm install -g localtunnel

Then create a tunnel to port 8000:

lt --port 8000

LocalTunnel will provide an HTTPS URL similar to:

https://your-generated-url.loca.lt

Keep the LocalTunnel process running while you're using the MCP server.

Note: LocalTunnel is useful for development and tutorials. For production, use a properly hosted HTTPS endpoint with authentication and appropriate security controls.

9. Connect the MCP Server to Claude

Now let's connect our MCP server to Claude.

In Claude, open:

Customize → Connectors

Then select:

Add a custom connector

Give the connector a name, for example:

Custom Todo

For the remote server URL, enter your LocalTunnel URL with the MCP endpoint:

https://your-generated-url.loca.lt/mcp

Then click Add.

Claude will try to connect to your MCP server.

If the connection succeeds, Claude should discover the two tools:

add_note
get_notes

Approve the tools so Claude can use them.

10. Test the `get_notes` Tool

Go back to the Claude chat and ask:

Show me my notes.

Claude should call the get_notes MCP tool.

You should get a response similar to:

Note #0: Remember to deploy server
Note #1: Eat Apple at 5 PM

The important part is that Claude is actually calling our MCP server to retrieve this information.

11. Test the `add_note` Tool

Now let's add a new note.

Ask Claude:

Add a new note: Research MCP authentication.

Claude should recognize that add_note is the appropriate MCP tool and call it with the provided text.

The server will add the note to our in-memory database.

Now ask Claude:

Show me my notes.

You should now see:

Note #0: Remember to deploy server
Note #1: Eat Apple at 5 PM
Note #2: Research MCP authentication.

We've successfully created an MCP server and connected it to Claude.

How the Request Flows

Here's what happens when you ask Claude to add a note:

You
 │
 │ "Add a new note..."
 ▼
Claude
 │
 │ MCP tool call
 ▼
HTTPS LocalTunnel URL
 │
 ▼
FastMCP Server
 │
 │ add_note()
 ▼
Mock Database
 │
 │ result
 ▼
Claude

This is the core idea behind MCP: an AI assistant can discover and interact with external tools through a standardized protocol.

From Mock Database to a Real Database

Our example uses a Python list:

notes = [...]

This is fine for learning, but the data will disappear when the server restarts.

A real application could use a persistent database such as PostgreSQL or SQLite:

FastMCP Server
      │
      ├── get_notes()
      │       │
      │       ▼
      │    Database
      │
      └── add_note()
              │
              ▼
           Database

You could also add more tools:

get_notes
add_note
update_note
delete_note
search_notes

At that point, you have the foundation for a proper AI-powered task or note management system.

Production Considerations

The LocalTunnel setup is great for experimentation, but you should take additional precautions before deploying an MCP server to production.

Consider:

Using a stable HTTPS domain
Adding authentication and authorization
Validating tool inputs
Using a persistent database
Adding logging and monitoring
Handling errors properly
Limiting access to sensitive tools
Protecting sensitive data
Adding rate limiting where appropriate

This is particularly important when an AI assistant has permission to perform actions on your behalf.

Conclusion

In this tutorial, we built a simple MCP server with FastMCP and connected it to Claude.

The process was:

Create a Python virtual environment.
Install FastMCP.
Create an MCP server.
Add get_notes and add_note tools.
Run the server on port 8000.
Expose the local server using LocalTunnel.
Add the server as a custom connector in Claude.
Use Claude to read and create notes.

The example is intentionally simple, but the same pattern can be used to expose APIs, databases, internal services, automation workflows, and other capabilities to AI assistants.

That's what makes MCP powerful: you can give an AI model access to capabilities, not just information.

If you're experimenting with MCP, start small. Build one useful tool, connect it to an MCP client, and then expand the server as your use case grows.

Day 2: Data Engineering vs Data Science vs Data Analytics

pawan deore — Fri, 12 Dec 2025 11:51:43 +0000

✅ Why Compare These Roles?

In modern data teams, Data Engineering, Data Science, and Data Analytics are three core pillars - but many people confuse them.

Knowing who does what:

Avoids misunderstandings in projects.
Helps you choose your career path wisely.
Makes collaboration smoother.

🗂️ The Big Picture

Role	Focus	Typical Tools
Data Engineer	Build & manage data pipelines, storage, & processing infrastructure.	SQL, Python, Spark, Hadoop, Airflow
Data Scientist	Develop models, run experiments, make predictions.	Python, R, TensorFlow, Scikit-learn
Data Analyst	Analyze data, build reports & dashboards, answer business questions.	SQL, Excel, Tableau, Power BI

👉 Key Difference:

Engineers build the highways.

Scientists build self-driving cars to run on them.

Analysts report on the traffic.

If you've been wanting to break into data engineering but don't know where to start, this guide gives you a simple, clear path to follow. Break Into Data Engineering: A Complete Roadmap for Beginners cuts through the noise and explains the essentials in a friendly, beginner-focused way across 15 comprehensive chapters and 190 pages. It's built to help you finally understand the field and know exactly what to learn next.

⚙️ What a Data Engineer Does

Main tasks:

Design data architecture (databases, data lakes, warehouses)
Develop, test, and maintain ETL/ELT pipelines
Integrate diverse data sources
Optimize storage & queries for performance
Monitor pipeline health & troubleshoot issues

Key goal: Deliver clean, structured, reliable data.

🔬 What a Data Scientist Does

Main tasks:

Explore & analyze large data sets
Build and test statistical & machine learning models
Perform A/B testing & experimentation
Interpret results and provide predictions
Communicate complex findings to stakeholders

Key goal: Turn data into actionable insights & predictive systems.

📊 What a Data Analyst Does

Main tasks:

Use SQL & BI tools to answer specific questions
Create dashboards and visual reports
Identify trends & patterns in historical data
Support decision-making with clear insights

Key goal: Help teams understand what happened and why.

🔑 Real-World Example

Example: E-commerce company

1️⃣ Data Engineer:

Sets up a pipeline to collect website clicks, purchases, and customer info.
Stores it in a data warehouse (e.g., Snowflake).

2️⃣ Data Scientist:

Uses that clean data to predict which customers are likely to churn.
Tests different retention strategies.

3️⃣ Data Analyst:

Builds daily reports showing sales trends, customer segments, and marketing campaign performance.

🎯 Key Takeaways for Day 2

✅ Data Engineers = Backbone: They build and maintain the data foundation.

✅ Data Scientists = Innovators: They create models that predict the future.

✅ Data Analysts = Explorers: They dig into past and present data to provide clear insights.

✅ These roles collaborate, not compete - each is vital for a modern data team.

🏃‍♂️ Action Step

Today's mini-task:

👉 Make a simple table:

One column: Your current skills
Second column: Engineer, Scientist, or Analyst? Tick what matches best - this helps you see where you fit now and where you want to grow!

Data Engineering in 30 Days - Day 2

pawan deore — Tue, 09 Dec 2025 12:26:27 +0000

Day 2: Data Engineer vs Data Scientist vs Data Analyst — What’s the Difference?

✅ Why Compare These Roles?

In modern data teams, Data Engineering, Data Science, and Data Analytics form three essential pillars — yet they’re often misunderstood or mixed up.

Understanding the differences helps you:

Avoid confusion in projects
Choose the right career path
Collaborate more effectively

If you’ve been wanting to break into data engineering but don’t know where to start, this guide gives you a simple, clear path to follow. Data Engineering: Complete Roadmap cuts through the noise and explains the essentials in a friendly, beginner-focused way across 15 comprehensive chapters and 190 pages.

🗂️ The Big Picture

Role Comparison Table

Role	Focus	Typical Tools
Data Engineer	Build & manage data pipelines, storage, and processing infrastructure	SQL, Python, Spark, Hadoop, Airflow
Data Scientist	Develop models, run experiments, make predictions	Python, R, TensorFlow, Scikit-learn
Data Analyst	Analyze data, build reports & dashboards, answer business questions	SQL, Excel, Tableau, Power BI

👉 Key Difference (Simple Analogy)

Data Engineers build the highways
Data Scientists build the self-driving cars
Data Analysts report on the traffic

📘 Want to Break Into Data Engineering?

If you’ve been wanting to break into data engineering but don’t know where to start, this guide lays out a super clean path:

Break Into Data Engineering: A Complete Roadmap for Beginners

A friendly, 190-page beginner-focused book covering the essentials in 15 structured chapters.

⚙️ What a Data Engineer Does

Main tasks:

Design data architecture (databases, data lakes, warehouses)
Develop, test, and maintain ETL/ELT pipelines
Integrate diverse data sources
Optimize storage & queries for performance
Monitor pipeline health & troubleshoot issues

Key goal:

Deliver clean, structured, reliable data.

🔬 What a Data Scientist Does

Main tasks:

Explore & analyze large datasets
Build and test statistical & machine learning models
Run A/B tests & experiments
Interpret and communicate findings
Provide predictions & insights

Key goal:

Turn data into actionable insights and predictive systems.

📊 What a Data Analyst Does

Main tasks:

Use SQL & BI tools to answer specific questions
Create dashboards & visual reports
Identify trends in historical data
Support decisions with clear insights

Key goal:

Help teams understand what happened and why.

🔑 Real-World Example: E-Commerce Company

1️⃣ Data Engineer

Builds pipelines to collect website clicks, orders, and customer data
Loads everything into a data warehouse (e.g., Snowflake)

2️⃣ Data Scientist

Uses the cleaned data to predict churn
Tests retention strategies

3️⃣ Data Analyst

Produces daily dashboards for sales, customer segments, and marketing performance

🎯 Key Takeaways for Day 2

Data Engineers = Backbone → They build and maintain the data foundation
Data Scientists = Innovators → They create models that predict the future
Data Analysts = Explorers → They uncover insights from past & present data
These roles collaborate, not compete — each is essential in modern teams

🏃‍♂️ Action Step

Today’s mini-task:

👉 Create a simple two-column table:

Your Current Skills	Engineer / Scientist / Analyst?

Mark where you fit today — this gives clarity on where you might want to grow!

End-to-End LLM Project Natural Language to SQL Application

pawan deore — Sat, 16 Aug 2025 13:37:03 +0000

FREE Course Link - https://www.udemy.com/course/end-to-end-llm-powered-natural-language-to-sql-application/

This course walks you through building a fully functional, LLM-powered application that transforms natural language into SQL queries and returns real-time results from a connected database. Designed for developers and AI enthusiasts, it focuses on real backend systems, LLM integration, and query logic—no deployment or theory-heavy content.

Introduction to LLM-Powered Applications

Learn the key concepts behind using Large Language Models like GPT for SQL generation. Understand the application architecture and flow: from natural language input to query execution and result rendering.

Setting Up Flask and Database Connections

Build the backend using Flask. Learn how to securely gather and store user connection info (host, port, user, password) and connect to PostgreSQL, MySQL, or SQLite using SQLAlchemy.

Integrating OpenAI API for SQL Generation

Use the OpenAI API to turn natural language queries into syntactically correct SQL. You'll design prompts, configure models, and handle API response parsing and error handling in your Python code.

Extracting Database Schema for Contextual Queries

Dynamically fetch and format schema information for more accurate SQL generation. Explore how to retrieve table names and column data types from your connected database to guide LLM output.

Improving Prompt Engineering and Query Accuracy

Refine how prompts are structured based on schema info and query intent. Add custom logic for PostgreSQL-specific behaviors like case sensitivity and identifier quoting.

Designing a Functional Web Interface with Flask

Create a simple HTML interface for inputting queries and displaying SQL results. Learn how to maintain session state, format results, and handle common user and database errors.

By the end, you'll have a complete AI-powered SQL interface, suitable for internal tools, BI assistants, or learning projects.

Data Science Projects You can start this weekend

pawan deore — Sun, 29 Jun 2025 15:32:44 +0000

Looking for some hands-on data science projects to sharpen your skills this weekend? Whether you're a beginner or an experienced practitioner, working on real-world projects is one of the best ways to learn. Below, we've curated 10 fantastic data science projects from a comprehensive list, spanning various domains like NLP, computer vision, time series forecasting, and MLOps.

Each project comes with a clear objective, relevant technologies, and a link to detailed instructions—so you can dive right in!

🔥 1. Digit Recognition using CNN for MNIST Dataset
Domain: Computer Vision / Deep Learning
Tech Stack: Python, TensorFlow/Keras, CNN

Why Try This?
The MNIST dataset is perfect for beginners to explore Convolutional Neural Networks (CNNs). You'll learn how to preprocess image data, build a CNN model, and evaluate its performance.

🔗 Project Link

📊 2. Time Series Forecasting with Facebook Prophet
Domain: Time Series Analysis
Tech Stack: Python, Facebook Prophet, Cesium

Why Try This?
Time series forecasting is crucial in finance, sales, and IoT. This project teaches you how to use Facebook Prophet, a powerful forecasting tool by Meta, to predict future trends.

🔗 Project Link

🤖 3. Text Classification with Transformers (RoBERTa & XLNet)
Domain: NLP / Transformers
Tech Stack: Python, Hugging Face, PyTorch

Why Try This?
Transformers like RoBERTa and XLNet dominate NLP tasks. This project walks you through fine-tuning these models for text classification, a skill useful in sentiment analysis, spam detection, and more.

🔗 Project Link

🛒 4. Market Basket Analysis using Apriori & FP-Growth
Domain: Recommendation Systems
Tech Stack: Python, Scikit-learn, Pandas

Why Try This?
Ever wondered how Amazon recommends products? This project uses association rule mining (Apriori & FP-Growth) to uncover product purchase patterns—essential for retail analytics.

🔗 Project Link

📈 5. Loan Default Prediction with Explainable AI
Domain: Finance / ML Interpretability
Tech Stack: Python, LightGBM, SHAP

Why Try This?
Banks need to understand why a loan might default. This project combines LightGBM with SHAP values to build a model that’s both accurate and interpretable.

🔗 Project Link

🏡 6. House Price Prediction with Regression Models
Domain: Regression / Predictive Analytics
Tech Stack: Python, Scikit-learn, Pandas

Why Try This?
A classic ML project! Predict house prices using linear regression, Ridge, and Lasso, while learning feature engineering and model evaluation.

🔗 Project Link

🚀 7. Deploy an ML Model with Streamlit & PyCaret
Domain: MLOps / Deployment
Tech Stack: Python, PyCaret, Streamlit

Why Try This?
Model deployment is a must-have skill. This project shows how to build and deploy an ML app quickly using PyCaret for automation and Streamlit for the UI.

🔗 Project Link

🎭 8. Fake News Detection with NLP & Deep Learning
Domain: NLP / Deep Learning
Tech Stack: Python, TensorFlow, LSTM

Why Try This?
Fake news is a growing problem. Learn how to classify news articles as real or fake using LSTM networks, a type of recurrent neural network (RNN).

🔗 Project Link

🛠️ 9. Build a CI/CD Pipeline for ML with Jenkins
Domain: MLOps / Automation
Tech Stack: Jenkins, Docker, Python

Why Try This?
CI/CD pipelines automate ML workflows. This project teaches you how to set up Jenkins for ML model testing and deployment, a valuable skill in production environments.

🔗 Project Link

🏎️ 10. Real-Time Streaming Pipeline with Spark & Kafka
Domain: Big Data / Real-Time Analytics
Tech Stack: PySpark, Kafka, AWS

Why Try This?
Real-time data processing is key in IoT and finance. This project guides you in building a Spark Streaming pipeline with Kafka for live data analysis.

🔗 Project Link

These projects cover diverse data science domains—from NLP and computer vision to MLOps and big data. Pick one that excites you and start coding this weekend!

💡 Pro Tip: If you're a beginner, start with MNIST Digit Recognition or House Price Prediction. If you're advanced, try Transformer-based NLP models or real-time Spark pipelines.

Happy coding! 🚀

Data Engineering in 30 Days: Day 1

pawan deore — Thu, 19 Jun 2025 03:05:29 +0000

✅ What is Data Engineering?

Data Engineering is the discipline focused on designing, building, and maintaining systems and pipelines that collect, store, process, and deliver data reliably and efficiently.

Key ideas:

It transforms raw data into usable data for analytics and machine learning.
It handles big volumes of data (terabytes to petabytes).
It ensures data is clean, consistent, and available to the right people and systems.

⚙️ Why is Data Engineering Important?

Without data engineering:

Data is messy, scattered, and unreliable.
Analysts and data scientists waste time cleaning data instead of extracting insights.
Companies struggle to make data-driven decisions.

With good data engineering:
✅ Business decisions are based on high-quality data.

✅ Data is fresh, trustworthy, and accessible.

✅ Complex analytics, dashboards, and ML models run smoothly.

In short: Data engineers build the foundation for all modern data-driven work.

If you’ve been wanting to break into data engineering but don’t know where to start, this guide gives you a simple, clear path to follow. Data Engineering: A Complete Roadmap cuts through the noise and explains the essentials in a friendly, beginner-focused way across 15 comprehensive chapters and 190 pages.

🔑 Typical Tasks of a Data Engineer

Here’s what data engineers do daily:

Build scalable pipelines: Automate the flow of data from multiple sources.
Integrate various systems: APIs, databases, IoT devices, and external feeds.
Clean and transform data: Fix errors, standardize formats, enrich data.
Design storage solutions: Databases, data lakes, and data warehouses.
Ensure security and governance: Control access and comply with privacy laws.
Monitor and maintain pipelines: Automate alerts and handle failures gracefully.

🗂️ Core Components in a Data Engineering Workflow

1️⃣ Data Sources:

APIs, transactional databases, server logs, sensors, third-party data.

2️⃣ Ingestion Layer:

Tools like Apache NiFi, Kafka, or custom scripts to bring in data.

3️⃣ Storage Layer:

Relational Databases (PostgreSQL, MySQL)
NoSQL Databases (MongoDB, Cassandra)
Data Warehouses (Snowflake, Redshift, BigQuery)
Data Lakes (AWS S3, Hadoop HDFS)

4️⃣ Processing Layer:

Batch processing — Spark, Hadoop
Streaming processing — Kafka Streams, Flink

5️⃣ Orchestration:

Workflow scheduling with Apache Airflow, Luigi.

6️⃣ Monitoring & Logging:

Set up alerts, logs, and dashboards to keep pipelines healthy.

🧰 Key Skills & Tools to Learn

Programming Languages:

Python: Most popular for scripting, ETL jobs, and working with frameworks.
SQL: Querying databases is a must-have skill.

Frameworks & Tools:

Apache Spark: For large-scale batch & stream processing.
Hadoop: Distributed storage & processing.
Apache Airflow: Schedule & orchestrate data workflows.
dbt (Data Build Tool): For managing transformations in the warehouse.

Cloud Platforms:

AWS (Glue, EMR, Redshift, S3)
Google Cloud (BigQuery, Dataflow)
Azure (Data Factory, Synapse)

📈 Example: How a Data Pipeline Works

Scenario: A company wants daily sales dashboards.

Pipeline Steps:

Extract: Pull raw sales transactions from the store’s POS database.
Transform: Clean data — fix missing values, convert currencies, join with product info.
Load: Store cleaned data into a data warehouse like Snowflake.
Serve: Analysts and BI tools (e.g., Tableau, Power BI) query this warehouse for reports.

✅ Automation ensures this happens daily with no manual work!

🎯 Key Takeaways for Day 1

✅ Data Engineering is the backbone of all analytics and AI work.

✅ It combines coding, system design, and an understanding of business data needs.

✅ Focus on building clean, reliable, and scalable pipelines.

✅ Start by mastering SQL, Python, and a basic ETL pipeline.

NumPy Interview Questions Practice Test Series

pawan deore — Sat, 17 May 2025 03:49:59 +0000

NumPy Interview Questions Practice Test Series - Complete List

Created Chrome Extension - Job Application Helper

pawan deore — Sun, 11 May 2025 09:10:45 +0000

Generates tailored responses to generic questions based on job descriptions and your resume

How It Works:

Enter Your Details – Provide your resume summary and the job description.
Generate Answers – Click the button to get AI-crafted responses in seconds.
Refine & Use – Copy the answers and personalize them further if needed.

Chrome Extension Link - Job Application Helper

DEV Community: pawan deore

Apache Spark Partitioning, Skew, and Broadcast Joins: A Practical Guide

Apache Spark Partitioning, Skew, and Broadcast Joins: A Practical Guide

Partitioning: Spark’s Unit of Parallel Work

How Joins Create Shuffles

A Runnable PySpark Example

When Broadcast Joins Help—and When They Do Not

Detecting and Fixing Data Skew

Let Adaptive Query Execution Help First

Use Key Salting for Severe Hot Keys

Production Pitfalls

Testing Partitioning and Join Behavior

Takeaway

Designing Airflow DAGs with Retries, SLAs, and Idempotency

Designing Airflow DAGs with Retries, SLAs, and Idempotency

The architecture: separate orchestration from data correctness

Model retries around failure types

Build an idempotent daily aggregation task

Example schema

Airflow DAG

Understand SLA timing before setting an SLA

Avoid the “idempotent SQL but changing source” trap

Test retries, writes, and schedules separately

Production pitfalls to plan for

Takeaway

CDC with Debezium and Kafka: A Source-to-Sink Walkthrough

CDC with Debezium and Kafka: A Source-to-Sink Walkthrough

The Architecture and Delivery Model

Prepare PostgreSQL for Logical Replication

Register the Debezium Connector

Generate Changes and Inspect the Kafka Records

Build a Small Idempotent Python Sink

Handle Ordering, Duplicates, and Schema Changes Deliberately

Production Pitfalls and Operational Controls

Test the Pipeline and Verify Recovery Behavior

Takeaway

Build an Incremental ETL Pipeline in Python Using PostgreSQL Watermarks

Build an Incremental ETL Pipeline in Python Using PostgreSQL Watermarks

Define the Watermark Contract First

Create Source, Target, and Pipeline-State Tables

Understand the Pipeline Architecture and Transaction Boundaries

Run the Incremental Python ETL Job

Verify Expected Results and Incremental Behavior

Production Pitfalls That Watermarks Do Not Solve Automatically

Test the Pipeline as a State Machine

Takeaway

Build an MCP Server with FastMCP and Connect It to Claude

🎥 Watch the Tutorial

What We'll Build

Prerequisites

1. Create a Virtual Environment

2. Install FastMCP

3. Create the MCP Server

4. Create the get_notes Tool

5. Create the add_note Tool

6. Configure the MCP Server

7. Run the MCP Server

8. Create an HTTPS Tunnel with LocalTunnel

9. Connect the MCP Server to Claude

10. Test the get_notes Tool

11. Test the add_note Tool

How the Request Flows

From Mock Database to a Real Database

Production Considerations

Conclusion

Day 2: Data Engineering vs Data Science vs Data Analytics

Data Engineering in 30 Days - Day 2

Day 2: Data Engineer vs Data Scientist vs Data Analyst — What’s the Difference?

✅ Why Compare These Roles?

🗂️ The Big Picture

Role Comparison Table

👉 Key Difference (Simple Analogy)

📘 Want to Break Into Data Engineering?

⚙️ What a Data Engineer Does

🔬 What a Data Scientist Does

📊 What a Data Analyst Does

🔑 Real-World Example: E-Commerce Company

1️⃣ Data Engineer

2️⃣ Data Scientist

3️⃣ Data Analyst

4. Create the `get_notes` Tool

5. Create the `add_note` Tool

10. Test the `get_notes` Tool

11. Test the `add_note` Tool