Send data to Deephaven from a Python client

This guide shows how to send data to Deephaven from an external Python application using pydeephaven and input tables. Input tables allow a client to add, update, and delete rows in a Deephaven table.

Note

If your data source can run directly on the Deephaven server, consider using server-side input tables or DynamicTableWriter instead. Server-side ingestion is generally more efficient because it avoids network overhead.

When to use client-side input tables

Use client-side input tables when:

Managing reference data: Upload configuration, lookup tables, or static datasets that may need updates.
Tracking state: Maintain the latest status per entity (e.g., device status, order state, user preferences).
Interactive editing: Allow users or external systems to add, update, or remove records.
Forwarding external data: Relay data from message queues, APIs, or sensors running in separate processes.

Basic pattern

The basic pattern for sending data from a Python client is:

Create an input table on the server using Session.input_table.
Prepare data in your client application.
Convert to PyArrow and upload with Session.import_table.
Add to the input table with InputTable.add. For keyed tables, this inserts new rows or updates existing ones.
Close the uploaded table with Table.close to release server resources and prevent memory leaks.

Repeat steps 2-5 as needed.

Complete example

The following example tracks device status using a keyed input table. Each device has a unique ID, and updates replace the previous status for that device.

import pyarrow as pa
from pydeephaven import Session
from datetime import datetime, timezone

# Define schema for device status
schema = pa.schema(
    [
        pa.field("DeviceId", pa.string()),
        pa.field("Status", pa.string()),
        pa.field("LastSeen", pa.timestamp("ns", tz="UTC")),
    ]
)

# Use context manager to ensure session is closed on success or failure
with Session() as session:
    # Create a keyed input table - DeviceId is the key
    # Adding a row with an existing DeviceId updates that row
    input_table = session.input_table(schema=schema, key_cols="DeviceId")

    # Bind to a name so it's visible in the UI (check http://localhost:10000)
    session.bind_table("device_status", input_table)

    def update_device(device_id: str, status: str):
        """Add or update a device's status."""
        pa_table = pa.table(
            {
                "DeviceId": [device_id],
                "Status": [status],
                "LastSeen": pa.array(
                    [datetime.now(timezone.utc)], type=pa.timestamp("ns", tz="UTC")
                ),
            }
        )

        uploaded = session.import_table(pa_table)
        try:
            input_table.add(uploaded)  # Inserts or updates based on DeviceId
        finally:
            uploaded.close()

    def remove_device(device_id: str):
        """Remove a device from the table."""
        keys = pa.table({"DeviceId": [device_id]})
        uploaded = session.import_table(keys)
        try:
            input_table.delete(uploaded)
        finally:
            uploaded.close()

    # Example usage
    update_device("sensor-001", "online")
    update_device("sensor-002", "online")
    update_device("sensor-003", "offline")

    # Update sensor-001's status (replaces the previous row)
    update_device("sensor-001", "maintenance")

    # Remove sensor-003
    remove_device("sensor-003")

Memory management

Each call to import_table creates a temporary table on the server. If you don't release these tables, they accumulate and consume server memory.

Always call close on uploaded tables after adding data to the input table. This releases the server-side export and marks the local object as closed. Use try/finally to ensure cleanup even if the add fails:

# Upload data
uploaded = session.import_table(pa_table)
try:
    # Add to input table
    input_table.add(uploaded)
finally:
    # Close the temporary table (even on failure)
    uploaded.close()

Input table types

The client supports three types of input tables:

Keyed

Rows are identified by key columns. Adding a row with an existing key replaces that row. Deletion is supported.

# Keyed by DeviceId
input_table = session.input_table(schema=schema, key_cols="DeviceId")

# With multiple key columns
input_table = session.input_table(schema=schema, key_cols=["DeviceId", "Region"])

With keyed tables, you can delete rows by providing just the key values:

# Delete rows by key
keys_to_delete = pa.table({"DeviceId": ["sensor-001", "sensor-003"]})
uploaded_keys = session.import_table(keys_to_delete)
try:
    input_table.delete(uploaded_keys)
finally:
    uploaded_keys.close()

Append-only

Rows are added to the end of the table. No key columns, no updates, no deletion. Use this when you need a simple log or event stream.

# Append-only (default when no key_cols specified)
input_table = session.input_table(schema=schema)

Blink

Rows are visible for only one update graph cycle, then disappear. Use this for event streams where you only need to process the latest batch.

# Blink table
input_table = session.input_table(schema=schema, blink_table=True)