Example 4: SFTP CSV -> PostgreSQL with Compute
Why this example
An external trading partner delivers a daily inventory file as a CSV dropped on an SFTP server. Internal reporting runs against PostgreSQL, so DataLug picks the file up, derives a few columns (full SKU, on-hand value), and lands the result in a Postgres table.
Helix Beverages context: two of Helix's largest distributors only support SFTP, and they drop nightly inventory files on their own schedule. This is how Operations gets those files merged into the PG inventory store before the 07:00 ops standup, without anyone babysitting the pipeline (partner schedule pain, named problem 3 in the Helix overview).
At a glance
- Connectors used: SFTP (file hub), PostgreSQL
- Nodes used: Start, Source, Compute, Transform, Load
- Schedule: daily at 06:30 UTC (after the partner's nightly upload completes)
- Direction: One-way (SFTP -> PostgreSQL)
Canvas
Start
Source: SFTP Get File
Compute: derive sku + value
Transform: align to PG schema
Load: PG inventory_daily
Step-by-step
| # | Node | Action | Key configuration | Output shape |
|---|---|---|---|---|
| 1 | Start | -- | -- | -- |
| 2 | Source | SFTP: Get File | Remote path /incoming/inventory/\{\{var.global.partnerCode\}\}_\{\{var.exec.runDate\}\}.csv. Parse CSV: header row, comma delimiter, UTF-8. | \{ sku_prefix, sku_suffix, qty_on_hand, unit_cost, warehouse_code \} |
| 3 | Compute | -- | Add sku = sku_prefix + "-" + sku_suffix, add on_hand_value = qty_on_hand * unit_cost. Keep all original columns. | \{ sku, sku_prefix, sku_suffix, qty_on_hand, unit_cost, warehouse_code, on_hand_value \} |
| 4 | Transform | -- | Drop sku_prefix and sku_suffix (no longer needed). Map to PG column names. | PG-shaped row |
| 5 | Load | PostgreSQL: Merge | Target inventory_daily, merge key (sku, warehouse_code, run_date). Mode: Merge. | -- |
Variations
- Compressed input -- the SFTP Get File action supports
.gz/.zipdecompression; set the file pattern and parse-format together. - Multiple partner files in one run -- replace Get File with List Directory, then iterate with the Loop node, then call Get File inside the loop body.
- Different file hub -- swap SFTP for Dropbox / OneDrive / Google Drive / S3 / Azure Blob; the rest of the canvas is identical.
- Reject bad rows -- add a Filter after Compute that requires
qty_on_hand > 0and route the DROP port to a Notification node for ops review.