node-red-contrib-parquet-file

node-red-contrib-parquet-file

A custom Node-RED node that reads and writes Apache Parquet files, modeled on the behavior of the core file node (filename field, msg-driven filename, overwrite/append actions, status dot).

Install

Copy this folder somewhere on the machine running Node-RED, then from your Node-RED user directory (usually ~/.node-red):

npm install /path/to/node-red-contrib-parquet-file

Restart Node-RED. A new parquet file node will appear in the storage category of the palette.

Alternatively, if you publish this to a private npm registry or git repo, npm install <package-or-git-url> works the same way.

Features

• Read Apache Parquet files
• Write Apache Parquet files
• Automatic schema detection
• Manual schema definition
• GZIP and Snappy compression
• Append and overwrite modes
• Row deduplication (upsert)
• Automatic directory creation
• Dynamic filenames via msg/flow/global context
• Safe concurrent writes

Usage

Write mode

Send msg.payload as either:

• a single object → one row, or
• an array of objects → many rows

Each object's keys become parquet columns.

• Overwrite: replaces the file with exactly the rows in this message.
• Append: reads the existing file (if any), merges in the new rows, and rewrites the whole file. Parquet's on-disk format can't be appended to in place — the footer is only written once, at close — so this is a full read-merge-rewrite, and the cost grows with the existing file's size (not just the new batch) — benchmarked at roughly linear growth, from ~30ms appending to an empty file up to ~2.4s appending 500 rows to a 200k-row file. Fine for periodic batches as long as the target file itself stays a bounded size; prefer rotating to a fresh file per hour/day over appending forever to one ever-growing file, and batch several polling cycles into one larger append rather than many small ones where practical, since each append call pays to rewrite the whole file regardless of how many new rows it carries.

Both Overwrite and Append write through a temp file and rename it into place atomically. A concurrent read of the same path (e.g. a dashboard node polling the file a scheduled writer is rewriting), or a process crash/power loss mid-write, only ever sees the fully-old file or the fully-new one — never a truncated/corrupt one.

Schema: "Auto-detect" infers each column's type by scanning every row in the batch (not just the first): numbers → INT64/DOUBLE, booleans → BOOLEAN, everything else (including Date objects, stored as ISO-8601 strings) → UTF8. On append, the column set auto-detect uses is the union of the existing file's columns and the new batch's columns — so a batch that happens to omit a column already present in the file (e.g. Influx omitting a null sensor reading) doesn't silently drop that column from the file's history. "Define manually" lets you fix column names/types so every message writes a consistent schema, and drops any payload fields not listed — recommended for anything landing in S3/a lakehouse (DuckLake, Athena, Spark), since a fixed schema avoids column-set or type drift across separate files, which auto-detect can't fully prevent on its own if you write many small files instead of appending to one.

Missing/null values: a field that's null, undefined, or simply absent from a row is written as a real parquet NULL, not coerced to 0 or "". This applies whether the value is missing on a normal write or backfilled onto older rows when auto-detect widens the schema on append — either way, "no reading" stays distinguishable from "reading of exactly zero" in downstream analytics.

Dedupe key validation: if any column named in Dedupe on doesn't match a real column in the file (a typo or case mismatch, e.g. Section vs section), the write is refused with an error listing the bad key and the known columns. Previously this failed silently and destructively: every row evaluates to the same key when the named column doesn't exist, so the merge would collapse the entire file down to a single row with no warning.

Known library issue: avoid the parquet TIMESTAMP_MILLIS type — parquetjs-lite has a bug reading it back on current Node.js (TypeError: Cannot convert a BigInt value to a number). Store timestamps as an ISO string (the auto-detect default) or as epoch milliseconds using INT64 instead.

Compression: defaults to GZIP. This matters a lot — uncompressed parquet carries enough per-file overhead (schema, headers, footer) that it can be larger than the equivalent CSV, especially with modest row counts. With GZIP, repetitive PLC/sensor data typically comes out to 5–20% of the CSV size. SNAPPY is available as a faster-but-larger alternative.

BigInt handling: parquetjs-lite returns INT64 columns as native JS BigInt on read, which breaks JSON.stringify and most downstream nodes. This node automatically converts them back to regular numbers (when safely representable) before setting msg.payload.

Dedupe on (append mode only): a comma-separated list of column names that uniquely identify a row, e.g. _time,section. If set, appending upserts on these keys instead of blindly concatenating — an incoming row with the same key as an existing one replaces it rather than duplicating it, and the file is kept sorted by the key. Essential if your upstream query can return overlapping data across runs (e.g. a fixed lookback window like "last 1 hour" on a more-frequent-than-hourly trigger), or if a run might get retried.

Invalid value handling: rows with NaN, unparseable strings, or non-finite values in numeric columns are dropped individually (not the whole batch) with a node.warn naming the row index and field. If every row in a message is invalid, the write fails with a clear error instead of a cryptic one. msg.rowsSkipped on the output reports how many rows were dropped.

Concurrency: writes/appends to the same file path are queued and run one at a time, in arrival order, even if two messages arrive close together (retries, overlapping schedules). Earlier versions had a race condition here that could corrupt output under concurrent writes to the same path — fixed as of v0.2.0. As of v0.4.0, both overwrite and append also write atomically (temp file + rename), so a read of the same path while a write is in flight — from another node instance, e.g. a dashboard — sees the old file cleanly rather than failing or reading a partial one, the file lock only serializes writers to the same path against each other.

If something else watches or syncs this file's directory (e.g. a flow that uploads new files to S3): as of v0.4.1, the write's temp file lives in a hidden .tmp subdirectory next to the target, not directly alongside it, specifically so a directory-scanning sync process doesn't pick up the in-progress temp file and move/delete it before this node's own rename completes (that raced and produced ENOENT ... rename ...tmp -> ...parquet errors in v0.4.0). This covers any sync tool that follows the standard convention of skipping dotfiles/dot-directories, which is most of them — but if yours is configured to sync everything unconditionally, also add an explicit exclude for .tmp/ (or dotfiles generally) in that tool's own config.

Read mode

Reads the whole file and sets msg.payload to an array of row objects, plus msg.rowCount.

Filename

Set a static path in the node, or switch the filename field to msg. / flow. / global. to source it dynamically. If left blank, msg.filename is used (same convention as the core file node).

Example: archiving InfluxDB data to partitioned Parquet

A common pattern: PLC/OPC UA data streams continuously into InfluxDB, and a separate scheduled job (every 30–60 min) queries Influx and archives it to partitioned Parquet on disk.

• Query InfluxDB on a fixed lookback window (e.g. Flux range(start: -1h)) on a timer trigger.
• Shape each point into a row object, e.g. { _time, section, TempPV, AgitRpmPV, ... }.
• Point this node at a partitioned path, e.g. /data/plant/date=20260713/vat.parquet, mode Write, action Append, with Dedupe on set to _time,section.

If your trigger interval is shorter than your lookback window (e.g. querying "last 1 hour" every 30 minutes as a safety margin against late-arriving points), every run will re-fetch data the previous run already wrote — the dedupe key is what keeps that from creating duplicate rows on every single run.

For high-frequency raw logging instead of a periodic archival job, consider writing to a fresh timestamped file per hour/day (action: create) instead of appending to one ever-growing file.

Version history

• 0.4.1 — fixed a production regression from v0.4.0's atomic-write change: the temp file used for the atomic rename lived directly alongside the target file, so an external process watching that directory (e.g. an S3 sync flow uploading new files) could pick up the in-progress temp file and move/delete it before this node's own rename ran, failing with ENOENT: no such file or directory, rename ...tmp -> ...parquet. The temp file now lives in a hidden .tmp subdirectory of the target's own directory instead — still the same filesystem, so the rename stays atomic, but outside what a typical directory listing/sync tool traverses by convention. See AUDIT.md for the repro and the remaining edge case (a sync tool that syncs literally everything, including dotfiles, still needs an explicit exclude on its own side).
• 0.4.0 — fixes from a full correctness/safety audit (see AUDIT.md for the full writeup with reproduction steps and benchmarks):
  • Dedupe key typos no longer silently destroy data. A mistyped/mismatched "Dedupe on" column name previously made every row evaluate to the same merge key, collapsing the whole file down to one row with no warning. Now validated up front — the write is refused with a clear error naming the bad key.
  • Overwrite is now crash-safe and race-safe. Overwrite previously opened and truncated the target file directly; a crash, OOM, or disk-full event partway through destroyed the previous good file with no recovery, and a concurrent read of the same path while a write was in progress would simply fail. Overwrite now writes through a temp file and renames atomically, same as append already did — a crash or a concurrent reader only ever sees the fully-old or fully-new file.
  • Missing/null values are stored as real parquet NULL, not fabricated as 0/"". Previously a sensor value that was null or absent was indistinguishable on read from an actual 0 reading, silently skewing any downstream average/sum/filter.
  • Auto-detect schema on append no longer drops columns. Column types/set are now inferred from the union of the existing file's columns and the new batch's, instead of the new batch alone — a batch missing a column the file already had (e.g. Influx omitting a null field) no longer erases that column from every existing row when the file is rewritten.
  • Removed a redundant second coercion pass on every row during writes (minor perf cleanup, no behavior change).
• 0.3.0 — fixed the root cause of a real production crash: auto-detect schema now scans every row in a batch to infer each column's type, instead of just the first row. Previously, a batch whose first row happened to have a whole-number value (e.g. an analog reading at exactly 3.0) would lock that column in as INT64, and any later row with a decimal (3.5) would crash the write with "The number 3.5 cannot be converted to a BigInt because it is not an integer". Also: non-integer values landing in an integer column (e.g. via manual schema) are now rounded instead of crashing, and INT64/INT32 overflow values are dropped as invalid (with a warning) instead of crashing the whole batch.
• 0.2.0 — dedupe-on-append (upsert by key columns), invalid-value validation (drops bad rows individually instead of crashing), fixed a race condition where concurrent writes to the same path could corrupt output, existing rows now re-coerced to current schema on append.
• 0.1.2 — GZIP compression by default (previously uncompressed, which could be larger than CSV), BigInt→Number conversion on read.
• 0.1.1 — auto-create missing directories before writing (fixes ENOENT on partitioned paths).
• 0.1.0 — initial release.

Dependencies

Uses parquetjs-lite — pure JS, no native build step, which keeps it easy to install alongside Node-RED.