Streaming

Streaming & Incremental

Match new records against existing data in real time. GoldenMatch supports single-record matching, micro-batch streaming, and CLI-based incremental matching.

StreamProcessor

Incremental record matching with immediate or micro-batch processing. Wraps match_one and add_to_cluster for continuous operation.

import goldenmatch as gm

processor = gm.StreamProcessor(existing_df, config)

# Process a single record immediately
matches = processor.process_record({"name": "John Smith", "zip": "10001"})

# Micro-batch mode: buffer records and process in batches
for record in incoming_records:
    processor.add_record(record)

# Flush the buffer
results = processor.flush()

Immediate mode

Each record is matched and clustered as it arrives:

processor = gm.StreamProcessor(df, config)
result = processor.process_record(new_record)
# result includes matches and updated cluster assignments

Micro-batch mode

Buffer records and process them together for better throughput:

processor = gm.StreamProcessor(df, config)
for record in batch:
    processor.add_record(record)
batch_results = processor.flush()

Incremental CLI

Match new CSV records against an existing base dataset:

goldenmatch incremental base.csv --new new_records.csv --config config.yaml

The incremental CLI handles exact and fuzzy matchkeys separately:

• Exact matchkeys: Polars join between new and base records (fast)
• Fuzzy matchkeys: match_one brute-force against the base (thorough)

Database watch mode

Continuously monitor a database table for new records and match them incrementally:

goldenmatch watch --table customers --connection-string "$DATABASE_URL" --interval 30

Daemon mode

Run watch as a background service with health endpoint and PID file:

goldenmatch watch --table customers --connection-string "$DATABASE_URL" --daemon

Daemon mode adds:

• HTTP health endpoint at /health
• PID file for process management
• SIGTERM handling for graceful shutdown

run_stream

Run a streaming pipeline programmatically:

import goldenmatch as gm

result = gm.run_stream(existing_df, config, new_records)