How a Python Processing Engine Speeds Time Series Data Processing
Time series data is information collected at specific time intervals: sensor readings, server performance or stock prices. Understanding how things change over time is essential to making decisions based on those trends. Collecting time series data helps teams monitor, forecast and respond to events quickly. But these actions are most valuable when they’re acted on in real time.
Real-time data fuels the systems we depend on: from industrial sensors to SaaS platforms. But the more data we collect, the harder it becomes to organize, analyze and act on it fast enough to make a meaningful difference.
Picture a logistics company tracking temperatures across thousands of refrigerated trucks. The sensors generate constant data, but alerts only trigger after batch processing — hours too late. Spoiled inventory, regulatory risks and missed deliveries follow.
The data is there, but legacy systems aren’t fast or flexible enough to keep up.
The Legacy Gap
Time series tools like Kapacitor and Flux played a key role in closing early processing gaps. Kapacitor supports real-time alerting and stream processing, while Flux tasks provide powerful, customizable transformations. As data sets grew in size and volume and systems became more complex, these tools needed updating. Teams needed a way to simplify processing, reduce latency and keep up with growing demands without adding more moving parts.
The Python Processing Engine
In response to the sheer volume of time series data, InfluxData introduced a built-in processing engine in InfluxDB 3 Core and Enterprise that runs Python code directly inside the database. This engine allows developers to process, analyze and act on time series data without relying on external tools or services.
Instead of sending data to another system for processing, developers can stay within the database to detect anomalies, enrich records with outside context or generate regular reports. Early uses include:
- A logistics team adds weather data to GPS signals from delivery trucks, improving safety and accuracy.
- An analyst sets up a weekly report on customer traffic patterns that’s automatically shared via Slack.
- A DevOps engineer watches for sudden changes in system metrics and sends alerts when thresholds are crossed.
These kinds of tasks once required multiple tools and complex workflows. Now, they can be handled inside the database using familiar Python libraries like Pandas and Plotly.
Why Python?
Python is simple to learn and widely used for data analysis and automation. Most teams already know it, and it works well with a range of tools, including cloud services and AI models.
Supporting Python inside the database gives developers a familiar way to build smarter workflows. And because the code runs where the data lives, businesses get faster results with less effort.
Even better, large language models (LLMs) make it easier to write and customize this code. With a few prompts, users can scaffold scripts that run inside the database without building everything from scratch.
The result? Faster development, fewer moving parts and more time spent solving problems — not stitching systems together.
Under the Hood: Plug-Ins and Triggers
The InfluxDB processing engine runs on plug-ins and triggers.
- Plug-ins: Custom Python scripts can access the full range of Python libraries to read, transform and write time series data. Each plug-in defines what you want to do with your data.
- Triggers: These are predefined events that tell the system when to run a Python script. There are three types:
- WAL flush: Runs every time the write-ahead log is flushed, typically once per second. Best for real-time transformations as data comes in.
- Scheduled tasks: Runs on a defined schedule using
cronsyntax. Ideal for daily reports, data cleanups or time-based aggregations. - On request: Executes when a GET or POST request is made to a custom endpoint under
/api/v3/engine. It’s great for one-time jobs or user-initiated actions.
This architecture provides clean, scalable automation that plugs into your existing workflows — no extra services or external systems required.
Fast Setup, Flexible Execution
Getting started with the processing engine is straightforward, and complete guidance is available in the documentation. You can download InfluxDB 3 Core or Enterprise here for free.
To begin, when launching your instance, specify a plug-in directory using the --plugin-dir flag. This tells the system where to find your Python scripts.
You can use example scripts provided in the GitHub repository or write your own from scratch. As more plug-ins are developed, this will become a growing ecosystem of reusable processing tools.
Once running, plug-ins can be configured with dynamic arguments — thresholds, filters, credentials and more — so you can fine-tune behavior without editing or redeploying code.
All Python scripts interact with a shared API that simplifies how you query, transform and write time series data. Everything executes inside the database, minimizing lag and complexity.
Want to enrich incoming data? Backfill historical records? Trigger custom alerts? It’s all possible with a few lines of Python.
Writing and Querying Data
Say you’re a renewable energy company tracking wind speedfrom sensors on remote turbines. You can’t wait around to react. You need to clean, label and write that data back for analysis and alerts. LineBuilder does the work for you. It takes the cleaned data, wraps it in line protocol and sends it right into the database — quick and clean, without extra steps.
|
1 2 3 4 5 |
line = LineBuilder("weather") .tag("location", "us-midwest") .float64_field("wind_speed", 22.5) .time_ns(1627680000000000000) influxdb3_local.write(line) |
This snippet creates a new measurement called weather, tags it with a location and logs a wind_speed value at a specific time. The write function commits that data to the database.
Sometimes. you need data from the past. You can use the query function to pull historical data directly from your plug-in. Since these queries are often analytical, the columnar storage engine in InfluxDB helps them run efficiently by quickly scanning and filtering large time ranges. This keeps the processing engine fast, even with heavy data loads.
|
1 2 3 4 |
results = influxdb3_local.query("SELECT wind_speed FROM weather WHERE time > now() - INTERVAL '10 minutes'") # Log the results to "info" for row in results: influxdb3_local.info(f"Sensor reading: {row}") |
This fetches all wind speed readings from the last 10 minutes and logs them for further processing. You might filter out anomalies, calculate rolling averages or flag unusual trends.
Logging Data
The processing engine includes built-in logging functions (info, warn and error) to track what’s happening as your code runs.
|
1 2 3 |
influxdb3_local.info("Wind speed normal") influxdb3_local.warn("Potential issue detected in weather data") influxdb3_local.error("Critical failure: unknown sensor issue") |
These logs persist in system tables and server logs, giving teams long-term visibility into system behavior and script performance.
Transforming Data
And because each script accepts arguments, you can keep things dynamic. Want to update a threshold without rewriting code? Just tweak the configuration:
|
1 2 3 4 5 6 7 |
def process_writes(influxdb3_local, table_batches, args=None): threshold = float(args["wind_threshold"]) for table_batch in table_batches: for row in table_batch["rows"]: wind_speed = row.get("wind_speed") if wind_speed > threshold: influxdb3_local.warn(f"High wind speed detected: {wind_speed} mph at {row['time']}") |
Want to change the threshold to 50 mph for an incoming storm? Just update wind_threshold in your configuration. There’s no need to modify or redeploy the script. That level of control helps teams stay agile, whether responding to weather, traffic surges or critical business events.
Why It Matters
InfluxDB 3 Core and Enterprise provides fast processing and flexible control with its built-in Python processing engine.
Embedding Python directly into a high-performance columnar database removes the need for external pipelines or added infrastructure. You can run custom logic, build alerts, generate reports and automate decisions — all where the data already lives.
That means businesses can ship insights faster, simplify their stack and respond to events in real time. Whether it’s generating a weekly report or detecting an outage before customers notice, teams can move from data to action without delays.
Get Started
The processing engine is now generally available in InfluxDB 3 Core and Enterprise. You can install it natively or run it via Docker. Download InfluxDB Core or Enterprise for free now. Start building using a sample plug-in or build custom logic to fit your needs.
Learn more with the InfluxData documentation. Connect with other developers in the InfluxData Discord.
