flowprep

Network telemetry → ML-ready canonical NetFlow parquet.

---

flowprep converts the network telemetry you actually have — packet captures, flow CSVs, vendor exports — into a single, clean, typed, unit-normalized parquet table that you can hand directly to a model, a notebook, or a data pipeline.

It is built and maintained by DeepTempo, where it is used in production as the ingestion front door for our LogLM: flow telemetry arriving from many different sources and formats is normalized through this same canonicalization layer, at scale, before it ever reaches inference. We open-sourced it because everyone working with network flow data ends up rebuilding this exact step, usually badly, usually more than once.

The problem

Every network-ML paper and every SOC data pipeline starts by solving the same unglamorous problem. Flow data arrives as:

• raw pcap/pcapng captures,
• CSV exports from research datasets (CIC-IDS, UNSW-NB15, …),
• nfdump/nfcapd binaries, Zeek logs, VPC flow logs,
• and a long tail of vendor formats, each with its own column names, duration units, and timestamp encodings.

The same field might be called src_ip, Source IP, srcaddr, ipv4_src_addr, or client_ip. A duration might be in seconds, milliseconds, microseconds, or nanoseconds — and nothing tells you which. A timestamp might be a datetime string, epoch seconds, or epoch nanoseconds.

Flow collectors (goflow2, nfdump) solve transport and storage, but they stop at JSON, protobuf, or their own binary formats. Feature-extraction tooling in the research world is fragmented and hard to reproduce. Nothing hands you a clean parquet table with one schema, ready for training or inference — including for the new generation of network foundation models.

flowprep is that missing step.

Quick start

# install from crates.io (Rust stable, no system dependencies — no libpcap, no JVM)
cargo install flowprep

# or build from source
cargo build --release

# or just run the demo against the bundled examples
./demo.sh

Five subcommands:

# 1. Raw packet captures -> bidirectional flow records
flowprep pcap capture.pcap flows.parquet

# 2. Any aliased flow table (CSV or parquet) -> the canonical schema
flowprep canonicalize cic_export.csv flows.parquet

# 3. OCSF Network Activity events (JSON/NDJSON) -> the canonical schema
flowprep ocsf network_activity.ndjson flows.parquet

# 4. nfdump/nfcapd binary flow files -> the canonical schema
flowprep nfcapd nfcapd.202401011200 flows.parquet

# 5. Inspect any parquet file from the terminal, no Python required
flowprep peek flows.parquet -n 20

The output is ordinary parquet. There are no bindings to install and no client library to learn — pandas, polars, DuckDB, Spark, and Arrow in any language read it natively. The file format is the API.

import pandas as pd
df = pd.read_parquet("flows.parquet")   # that's the whole integration

What it does

Field alias resolution

flowprep knows 100+ column spellings observed in the wild — CICIDS's Total Fwd Bytes, nfdump's in_bytes, IPFIX's l4_src_port, Zeek-style names, and many vendor variants — and maps them all onto one canonical schema. The alias map is not hard-coded: it is loaded at compile time from a declarative schema file (schemas/netflow/v1/schema.json), the same artifact DeepTempo's production ingestion uses. Adding support for a new vendor's column names is a JSON edit, not a code change.

Unit and encoding normalization

• Durations arrive in seconds, milliseconds, microseconds, or nanoseconds; the unit is inferred from the source column's name (duration_ms, flow_duration_microseconds, …) and everything lands as float64 seconds.
• Timestamps arrive as datetime strings, typed timestamps, or epoch values at any precision; epoch precision is inferred from magnitude and everything lands as int64 epoch microseconds.
• Protocols arrive as IANA numbers or names (tcp, udp, icmp); names are mapped to numbers.

OCSF Network Activity events

OCSF (the Open Cybersecurity Schema Framework) is a standard, not a vendor dialect, so flowprep reads its Network Activity events (class_uid 4001) directly rather than through the alias map. Endpoints come from src_endpoint/dst_endpoint, byte and packet counts from traffic (with a top-level bytes_from_client/bytes_from_server fallback), and time/ duration are converted from the OCSF millisecond convention to the canonical epoch-microsecond timestamp and float-second duration. Only flow-close events (activity_name "Closed" or activity_id 2) are kept, since those carry the final byte totals. Input may be NDJSON (one event per line), a JSON array, or a single object. Events are deserialized into a typed view of the standard's shape rather than navigated as loose JSON, and malformed records or close events missing required fields are reported as errors rather than silently dropped — partial or empty output never looks like success.

nfdump/nfcapd binary flow files

nfcapd files are nfdump's native on-disk format: a typed binary container of already-aggregated flow records (NetFlow v5/v9, IPFIX, sFlow), not packets. So unlike the pcap reader there is no 5-tuple aggregation — each stored record maps to one canonical flow. flowprep reads both the V1 "common" record written by nfdump 1.6.x and the V2 "v3" extension record written by 1.7.x, and transparently decompresses LZO/BZ2/LZ4/ZSTD/uncompressed data blocks. Endpoints, ports, and protocol map straight across; nfdump's millisecond-epoch times become the canonical epoch-microsecond timestamp and float-second duration. Biflow exporters carry a reverse counter, which maps to bwd_bytes/bwd_pkts; single-counter records zero-fill the backward side, the same convention the CSV and OCSF readers use. There is no nfdump CLI dependency — the container is parsed in-process, so flowprep stays a single static binary you can copy to a sensor box. (The very old pre-1.6 "common v0" record type is not decoded; such files convert to zero flows and are reported as an error rather than a silent empty file.)

Bidirectional flow aggregation (pcap)

Packets are grouped by a direction-normalized 5-tuple, so both halves of a conversation aggregate into a single flow record with separate forward/backward byte and packet counters. Flows split on a 60s idle timeout and a 1h maximum duration. The reader streams pcap and pcapng, keeps constant memory on the packet path, and is robust to the slightly-out-of-order packets real captures contain.

Label passthrough

Ground-truth columns (label, attack, attack_type, …) survive canonicalization unchanged, so labeled research datasets stay labeled — convert once, train immediately.

Canonical schema

field	type	notes
timestamp	int64	flow start, epoch microseconds
src_ip / dest_ip	string
src_port / dest_port	int32
fwd_bytes / bwd_bytes	int64	bwd zero-filled for single-counter sources
fwd_pkts / bwd_pkts	int64	nullable
flow_dur	float64	seconds
protocol	int32	IANA number; names auto-mapped

Plus any passthrough label columns present in the source.

Example: a real research dataset

The repo ships a ~100k-row slice of CIC-IDS-2017 with its original quirks intact — aliased packet columns (total_fwd_pkts), typed datetime timestamps, and attack labels:

$ flowprep canonicalize examples/cic2017_sample.parquet /tmp/cic.parquet
Wrote 101094 flows to /tmp/cic.parquet        # ~60 ms

$ flowprep peek /tmp/cic.parquet -n 3
+------------------+---------------+--------------+----------+-----------+-----------+-----------+----------+----------+----------+----------+--------+-------+
| timestamp        | src_ip        | dest_ip      | src_port | dest_port | fwd_bytes | bwd_bytes | fwd_pkts | bwd_pkts | flow_dur | protocol | attack | label |
+------------------+---------------+--------------+----------+-----------+-----------+-----------+----------+----------+----------+----------+--------+-------+
| 1488876958000000 | 8.254.250.126 | 192.168.10.5 | 80       | 49188     | 12        | 0         | 2        | 0        | 4e-6     |          | benign | 0     |
| 1488877019000000 | 192.168.10.9  | 192.168.10.3 | 1056     | 88        | 2812      | 2820      | 7        | 4        | 0.000655 |          | benign | 0     |
| 1488877062000000 | 192.168.10.17 | 192.168.10.3 | 35504    | 88        | 3150      | 3152      | 10       | 6        | 0.001122 |          | benign | 0     |
+------------------+---------------+--------------+----------+-----------+-----------+-----------+----------+----------+----------+----------+--------+-------+

The canonical output of this exact command is published as a ready-to-use dataset on the Hugging Face Hub — DeepTempo/cic-ids-2017-flowprep (101,094 labeled flows; drop-in for pandas / polars / datasets).

Performance

The per-packet decode path is exactly where interpreted languages pay the serialization tax, and it's why flowprep is a compiled tool. On a 500k-packet capture (Apple Silicon laptop, single thread):

implementation	throughput	wall time
flowprep (Rust)	~1.4M packets/s	0.35 s
equivalent Python (dpkt + pyarrow)	~78k packets/s	6.4 s

That's an 18.5x difference producing byte-identical flow output (tests/bench_pcap.py reproduces the measurement). Canonicalizing the 101k-row CIC sample takes ~60 ms end to end, including parquet read and zstd-compressed write.

Design notes

• Apache Arrow is the data plane. flowprep is built on arrow-rs and the parquet crate; columnar data goes from reader to writer without detours through row-by-row object representations.
• Schema as data, not code. The canonical field set, alias mappings, and unit-detection rules live in one JSON artifact embedded at compile time. Tools and pipelines in other languages can consume the same file.
• Single static binary. No venv, no JVM, no libpcap, no runtime dependencies. cargo build --release produces one file you can copy to an air-gapped sensor box.

Production use at DeepTempo

DeepTempo builds LogLM — a log language model for network security that turns flow telemetry into embeddings and incident classifications. Customer flow data arrives in wildly different shapes (cloud flow logs, IPFIX exporters, pcaps, SIEM exports), and this canonicalization layer is how all of them converge to the one schema the model consumes, at production scale, ahead of inference. flowprep is that layer, maintained in the open: if it mangles a format you care about, the fix benefits our pipeline and yours equally.

Roadmap

• Zeek conn.log reader
• IPv6 flow-tuple test coverage and pcapng per-interface timestamp resolutions
• More published canonical-parquet research datasets (CIC-IDS-2017 is live)

Contributions welcome — especially "here is a flow export flowprep can't parse" issues with a small sample attached.

Development

cargo build --release

# end-to-end tests (python harness generates fixtures; needs dpkt + pyarrow)
python3 -m venv .venv && .venv/bin/pip install dpkt pyarrow
.venv/bin/python tests/test_e2e.py

# throughput benchmark
.venv/bin/python tests/bench_pcap.py

Release

Releases are tag-driven. Pushing a v* tag builds Linux, macOS, and Windows binaries, attaches them to a GitHub release, and publishes the crate to crates.io using CARGO_REGISTRY_TOKEN.

Acknowledgments

The nfdump/nfcapd binary reader (src/nfdump/) is a small, purpose-built parser that decodes only the flow-bearing fields flowprep needs. Its byte layouts are derived from nfdump's on-disk format (phaag/nfdump) and the nfdump crate (ISC, © 2023 Mark King). LZO1X block decompression uses the pure-Rust, MIT-licensed lzokay-native (a clean-room lzokay port) rather than the GPL system liblzo2, so flowprep stays a single static binary. The reader validates every length before use and reports truncation or corruption as an error rather than a silent short read. See src/nfdump/mod.rs.

The bundled example slice derives from the CIC-IDS-2017 dataset: Sharafaldin, Lashkari & Ghorbani, "Toward Generating a New Intrusion Detection Dataset and Intrusion Traffic Characterization", ICISSP 2018 (Canadian Institute for Cybersecurity, University of New Brunswick). It is included solely as a small conversion example; for research use, obtain the full dataset from UNB CIC and cite accordingly.

License

Apache-2.0 © DeepTempo