Data Transformations with Pandas

This article covers a common transformation pattern: read data from ClickHouse into a Pandas DataFrame, transform and validate it in Python, then write the prepared result back to an analytical layer.

The goal is not to build one particular application. It is to make common Pandas operations easier to read and reuse in data work.

Mental model#

ClickHouse Raw      ↓clickhouse-connect      ↓Pandas DataFrame      ↓Transformations      ↓Validation / Filtering      ↓ClickHouse Staging

ClickHouse provides the source and destination storage. clickhouse-connect handles the Python connection. Pandas provides an in-memory table for inspection and transformation.

The DataFrame is the working surface between the database and the prepared analytical layer. The database remains responsible for persistence; Python and Pandas apply the transformation logic.

Connecting to ClickHouse#

The clickhouse-connect package provides a Python client for ClickHouse:

from clickhouse_connect import get_client
client = get_client(    host="clickhouse",    port=8123,    username="default",    password="password",)

Use the hostname configured for your environment. In a real application, credentials and connection settings should come from environment variables or application configuration rather than being hardcoded in source code.

Reading ClickHouse data into Pandas#

query_df() runs a query and returns its result directly as a Pandas DataFrame:

df = client.query_df("""    SELECT *    FROM raw.orders""")

Before transforming anything, inspect the data you actually received:

df.head()df.columnsdf.dtypesdf.shapedf.isna().sum()

These checks reveal the first rows, available columns, data types, row and column counts, and missing values. Transformation code depends on the schema, so inspecting it first prevents many avoidable errors.

Understanding the DataFrame#

A DataFrame is a table with labeled columns and an index. Selecting one column usually returns a Series:

amounts = df["amount"]

Selecting multiple columns returns another DataFrame:

orders = df[["order_id", "customer_id", "amount"]]

The distinction matters because Series and DataFrames support related but different operations. It also helps explain the shape of results returned by groupby(), agg(), transform(), and apply().

Working with datetime columns#

Database timestamps may arrive in different formats or may be read as plain strings. Convert them before sorting, filtering, grouping, or calculating durations:

import pandas as pd

df["created_at"] = pd.to_datetime(    df["created_at"],    errors="coerce",    utc=True,)

errors="coerce" converts values that cannot be parsed into NaT. The resulting missing timestamps can then be validated explicitly:

invalid_created_at_mask = df["created_at"].isna()invalid_timestamps = df[invalid_created_at_mask]

With a datetime Series, the .dt accessor exposes date and time components:

df["created_day"] = df["created_at"].dt.floor("D")df["created_hour"] = df["created_at"].dt.hour

Elapsed time between rows#

.diff() on a datetime Series returns, for each row, the gap to the previous row as a Timedelta. .dt.total_seconds() turns each gap into a float number of seconds:

track = df.sort_values("timestamp")
dt_seconds = track["timestamp"].diff().dt.total_seconds()

Two things to keep in mind:

• It is order-dependent. diff() compares each row to the one above it, so sort by the timestamp first.
• The first row has no predecessor, so its value is NaN. Fill or drop it before using the result in arithmetic.

When rows belong to several entities, compute the gap within each group so the first row of one entity is not compared to the last row of another:

df = df.sort_values(["track_id", "timestamp"])
df["gap_seconds"] = (    df.groupby("track_id")["timestamp"]    .diff()    .dt.total_seconds())

gap_seconds is then the time since the previous record for the same track_id — the denominator for a rate, or an input to a speed calculation (distance / gap_seconds).

Datetime values can be filtered with Pandas timestamps and sorted in the same way as other columns:

start_ts = pd.Timestamp.now(tz="UTC") - pd.Timedelta(days=7)
recent_orders = df[df["created_at"] >= start_ts]recent_orders = recent_orders.sort_values("created_at")

Use one timezone convention throughout the DataFrame. Mixing timezone-aware and timezone-naive values can make comparisons fail or produce ambiguous results. Converting at the boundary with utc=True is one practical way to keep the column consistent.

Filtering rows#

The simplest filter keeps rows where a condition is true:

df = df[df["amount"] > 0]

For multiple conditions, use & for AND and | for OR. Put each comparison inside parentheses:

df = df[    (df["amount"] > 0)    & (df["status"] == "completed")]

Each comparison returns a boolean Series. Pandas combines those values row by row to decide which records remain. Parentheses are required because Python's operator precedence would otherwise evaluate the expression incorrectly.

Creating derived columns#

Prefer vectorized column operations for calculations that apply independently to every row:

df["total"] = df["price"] * df["quantity"]

Comparisons return a boolean Series. Convert it to integers when a downstream system expects a numeric flag:

LIMIT = 1000
df["is_large_order"] = (    df["total"] > LIMIT).astype(int)

Vectorized expressions operate on whole columns and are usually clearer and faster than calling a Python function once per row.

GroupBy fundamentals#

The mental model for groupby() is:

DataFrame    ↓split rows into groups by a key    ↓perform an operation per group    ↓combine the results

Assume several rows can belong to the same customer. The first and last row for each customer can be selected like this:

first_order = df.groupby("customer_id").first()last_order = df.groupby("customer_id").last()

After groupby("customer_id").first():

• customer_id becomes the index
• one row remains per customer
• values come from the first row of each group

last() follows the same shape: one row per customer, with values from the last row of each group. The meaning of “first” and “last” depends on the row order, so sort explicitly when the order matters:

df = df.sort_values(["customer_id", "created_at"])first_order = df.groupby("customer_id").first()last_order = df.groupby("customer_id").last()

Aggregating groups#

Aggregations reduce many rows into values that describe each group:

customer_stats = (    df.groupby("customer_id")["created_at"]    .agg(["min", "max"]))

Conceptually, the result has one row per customer:

customer_id | min                  | max------------ | -------------------- | --------------------101          | first observed time  | last observed time102          | first observed time  | last observed time103          | first observed time  | last observed time

Named aggregations make a multi-column summary easier to read:

stats = df.groupby("customer_id").agg(    first_seen=("created_at", "min"),    last_seen=("created_at", "max"),    total_spent=("amount", "sum"),    order_count=("order_id", "count"),)

The result still has one row per customer. agg() changes the row cardinality from one row per input record to one row per group.

Mapping group results back to rows#

Sometimes a group-level result is needed on every original row. The pattern is:

1. Compute one value per group.
2. Get a Series indexed by the group key.
3. Map those values back onto every row.

For example:

customer_total = (    df.groupby("customer_id")["amount"]    .sum())
df["customer_total"] = (    df["customer_id"]    .map(customer_total))

The customer_total Series conceptually looks like this:

customer_id101    1250.0102     420.0103     890.0

df["customer_id"].map(customer_total) looks up each customer ID in that Series. Every input row receives the total belonging to its customer, so a group-level value becomes a row-level column.

The same pattern can attach the first and last observed times:

customer_time = (    df.groupby("customer_id")["created_at"]    .agg(["min", "max"]))
df["first_seen"] = (    df["customer_id"]    .map(customer_time["min"]))
df["last_seen"] = (    df["customer_id"]    .map(customer_time["max"]))

customer_time["min"] and customer_time["max"] are Series indexed by customer_id. map() uses the value in each row's customer_id column as a lookup key and aligns the result to the original DataFrame.

Using groupby().apply()#

apply() runs a custom Python function for each group:

def classify_customer(group):    total = group["amount"].sum()
    if total > 1000:        return "high_value"
    return "regular"

customer_category = (    df.groupby("customer_id")    .apply(classify_customer))
df["customer_category"] = (    df["customer_id"]    .map(customer_category))

apply() is flexible because the function can contain custom Python logic and return a value derived from the complete group. It is often slower and harder to reason about than vectorized operations or built-in aggregations.

Prefer agg(), transform(), and built-in vectorized operations when they express the logic clearly.

agg() vs. transform()#

agg() returns one result per group:

totals = df.groupby("customer_id")["amount"].sum()
df["customer_total"] = (    df["customer_id"]    .map(totals))

transform() returns a result aligned to the original DataFrame rows:

df["customer_total"] = (    df.groupby("customer_id")["amount"]    .transform("sum"))

Both examples can solve the same problem. Use transform() when the group result should be repeated for every original row. Use agg() when a compact one-row-per-group result is what you need, or when you want to map it explicitly.

Boolean masks#

A mask is a Series of True and False values. df[mask] keeps rows where the mask is True, and ~ reverses it:

high_value_mask = df["total"] > 1000normal_mask = ~high_value_mask
high_value_orders = df[high_value_mask]df = df[normal_mask]

Masks can combine several conditions:

mask = (    (df["total"] > 100)    & (df["status"] == "completed"))
filtered = df[mask]

Boolean masks are useful for validation, filtering, and splitting a DataFrame into separate outputs without writing an explicit loop.

Missing values and type conversion#

Mapping may produce missing values when an input key has no matching group result. A method chain can fill the missing values and enforce the expected type:

df["customer_total"] = (    df["customer_id"]    .map(customer_total)    .fillna(0.0)    .astype(float))

Read the chain as three operations:

map(...)→ attach the group result
fillna(0.0)→ replace missing values
astype(float)→ guarantee the expected type

When a chain becomes hard to read or debug, break it into intermediate variables:

values = df["customer_id"].map(customer_total)values = values.fillna(0.0)values = values.astype(float)
df["customer_total"] = values

This is often better while learning, debugging, or adding validation between steps.

Incremental loading#

An incremental read loads only new or changed records instead of reading the entire Raw table every time.

First, find the latest processed value in the Staging table:

result = client.query("""    SELECT maxOrNull(created_at)    FROM staging.orders""")
last_ts = result.result_columns[0][0]

result_columns exposes the query result as a list of columns, and the first column's first value is the returned maximum. If the table is empty, maxOrNull() returns None.

The next query can choose a full read for the first run or only newer rows for later runs:

if last_ts is None:    sql = """        SELECT *        FROM raw.orders    """else:    sql = f"""        SELECT *        FROM raw.orders        WHERE created_at > '{last_ts}'    """
df = client.query_df(sql)

The mental model is:

Staging   ↓find the latest processed value   ↓Raw   ↓read only newer rows

Incremental loading improves efficiency but adds correctness risks:

• duplicate timestamps can cause missed or repeated rows
• late-arriving records may fall behind the checkpoint
• timezone mismatches can shift the boundary
• interpolated values can make query construction unsafe

Production code should use safer parameter binding or query parameters where supported instead of blindly interpolating external values. The boundary strategy should also be documented and tested.

Handling an empty DataFrame#

Scheduled jobs should handle a run with no new records explicitly:

if df.empty:    return

This prevents later transformations or inserts from doing unnecessary work when the incremental query returns no rows.

Writing a DataFrame back to ClickHouse#

insert_df() writes a DataFrame to a ClickHouse table. Select the columns explicitly before inserting:

columns = [    "order_id",    "customer_id",    "created_at",    "total",    "customer_total",    "is_large_order",]
client.insert_df(    "staging.orders",    column_names=columns,    df=df[columns],)

Explicit selection is safer than inserting the entire DataFrame. It protects the target schema from temporary inspection columns, unexpected source fields, or columns in the wrong order.

Full transformation example#

The following function connects to ClickHouse, reads completed orders, derives row-level and customer-level values, and writes a defined output schema:

import pandas as pdfrom clickhouse_connect import get_client

def transform_raw_orders_to_staging():    client = get_client(        host="clickhouse",        port=8123,        username="default",        password="password",    )
    df = client.query_df("""        SELECT            order_id,            customer_id,            created_at,            price,            quantity,            status        FROM raw.orders        WHERE status = 'completed'    """)
    if df.empty:        return
    df["total"] = df["price"] * df["quantity"]
    customer_time = (        df.groupby("customer_id")["created_at"]        .agg(["min", "max"])    )
    df["first_seen"] = (        df["customer_id"]        .map(customer_time["min"])    )
    df["last_seen"] = (        df["customer_id"]        .map(customer_time["max"])    )
    df["customer_total"] = (        df.groupby("customer_id")["total"]        .transform("sum")    )
    LARGE_ORDER_LIMIT = 1000
    df["is_large_order"] = (        df["total"] > LARGE_ORDER_LIMIT    ).astype(int)
    columns = [        "order_id",        "customer_id",        "created_at",        "total",        "first_seen",        "last_seen",        "customer_total",        "is_large_order",    ]
    client.insert_df(        "staging.orders",        column_names=columns,        df=df[columns],    )

The operations have distinct purposes:

1. The query selects only the source fields needed by the transformation.
2. The empty check makes a scheduled no-op safe.
3. price * quantity creates a value for each order.
4. groupby().agg() computes the first and last time per customer.
5. map() attaches those group-level values to every order row.
6. transform("sum") repeats each customer's total on that customer's rows.
7. The comparison creates a numeric flag for large orders.
8. Explicit columns define the schema written to Staging.

The code is a complete example of a transformation function, not a universal template. Production code should also externalize connection settings, add data quality checks, and define the rerun behavior for the target table.

How to read complex Pandas code#

Read a Pandas chain from the inside out and from left to right. For example:

df["customer_total"] = (    df["customer_id"]    .map(customer_total)    .fillna(0.0)    .astype(float))

Break it down as follows:

1. Select the customer_id Series.
2. Map each ID to a precomputed customer total.
3. Replace missing values with 0.0.
4. Convert the result to float.
5. Assign the result to a new DataFrame column.

For this grouped expression:

stats = (    df.groupby("customer_id")["created_at"]    .agg(["min", "max"]))

The reading process is:

1. Start with the DataFrame.
2. Group rows by customer_id.
3. Select the created_at column inside each group.
4. Compute min and max for every group.
5. Return an aggregated DataFrame with one row per group.

This inside-out method works for most Pandas chains: identify the object at each step, note its shape, then check how the next method changes it.

Mental model for Pandas#

df["column"]→ Series
df[["a", "b"]]→ DataFrame
df.groupby("key")→ grouped object
.groupby(...).agg(...)→ usually one row per group
.groupby(...).transform(...)→ result aligned with original rows
.groupby(...).apply(...)→ run custom logic per group
Series.map(...)→ look up values using Series elements as keys
df[mask]→ filtered DataFrame

Keeping both the object type and row cardinality in mind makes non-trivial transformations much easier to follow.

Airflow#

The transformation function can later be executed by an orchestration system such as Airflow:

Airflow DAG     ↓Python transformation task     ↓ClickHouse Raw     ↓Pandas     ↓ClickHouse Staging

Airflow coordinates when the function runs and what it depends on. The Python function contains the transformation logic. Airflow and DAG design are covered in the Orchestration section.

Common mistakes#

Forgetting that groupby().agg() changes row cardinality#

Aggregation returns one result per group, not one result per original row. Use transform() or map() when the value must be attached back to every row.

Assuming groupby().apply() preserves the original DataFrame shape#

apply() returns whatever the custom function produces. Inspect its result before assigning it to a column or writing it to a table.

Using apply() when vectorized operations are simpler#

A custom function may be slower and harder to test than a column expression, built-in aggregation, or transform().

Forgetting that group keys often become the index#

Results from groupby() and agg() commonly use the group key as an index. Check the index before mapping or inserting the result into another system.

Mapping from an incompatible Series#

map() expects lookup values indexed by keys compatible with the Series being mapped. A mismatch produces missing values or incorrect results.

Ignoring missing values after map()#

Unmatched keys become missing values. Decide whether to fill, reject, or explain them before writing the output.

Forgetting type conversion before database insertion#

Pandas may infer a type that does not match the target schema. Inspect and cast columns explicitly when the destination requires a specific type.

Mutating a DataFrame without checking the resulting schema#

Temporary columns, renamed fields, and inferred types can accidentally reach the destination. Select and validate the final column set before insertion.

Using a full reload when incremental loading is enough#

Full reloads can become expensive as the Raw table grows. Use incremental reads when the boundary, state, and late-record behavior are well defined.

Constructing unreadable one-line chains#

Long chains hide intermediate values and make failures difficult to locate. Use named intermediate variables when a transformation needs explanation, validation, or debugging.

Quick reference#

Which one should I use?#

A practical Pandas transformation usually becomes easier to maintain when each operation has a clear shape, input, and output.

See also#