dataintermediate20 min setupby Mantle

Nightly CSV feed to SQL

Download a CSV export every night, parse it, normalise the columns with the Process engine and upsert into a table.

Request this template in a demo Talk to sales

Drag to pan, pinch to zoom. This is the real graph the template installs.

Purpose

Download a CSV export every night, parse it, normalise the columns with the Process engine and upsert into a table.

How it works

What it does

Cron 0 2 * * * → download the CSV → Parse CSV (headers, rows, row_count).
Normalise columns runs Process-engine cards (trim/lowercase email, title-case name, literal source) with a required-email rule that aborts on bad rows.
Upsert customers writes the shaped record keyed on email.

Nodes 5

01Nightly at 02:00trigger.scheduleTrigger
02Download CSVhttp.requestHTTP
03Parse CSVfiles.parse_csvData
04Normalise columnsdata.processData
05Upsert customersdb.upsertDestination

View template JSON

{
  "format": "mantle-workflow/v1",
  "id": "data-csv-sync-to-sql",
  "name": "Nightly CSV feed to SQL",
  "graph": {
    "nodes": [
      {
        "id": "schedule",
        "type": "trigger.schedule",
        "label": "Nightly at 02:00",
        "config": {
          "cron_expr": "0 2 * * *",
          "timezone": "UTC"
        },
        "worker_type": "generic-worker",
        "position": {
          "x": 60,
          "y": 220
        }
      },
      {
        "id": "fetch_csv",
        "type": "http.request",
        "label": "Download CSV",
        "config": {
          "method": "GET",
          "url": "https://example.com/exports/customers.csv",
          "timeout_seconds": 60
        },
        "worker_type": "generic-worker",
        "position": {
          "x": 380,
          "y": 220
        }
      },
      {
        "id": "parse_csv",
        "type": "files.parse_csv",
        "label": "Parse CSV",
        "config": {
          "file_content": "{{ steps.fetch_csv.body }}",
          "has_header": true
        },
        "worker_type": "generic-worker",
        "position": {
          "x": 700,
          "y": 220
        }
      },
      {
        "id": "process_rows",
        "type": "data.process",
        "label": "Normalise columns",
        "config": {
          "operations": [
            {
              "id": "op_email",
              "kind": "map",
              "sources": [
                "email"
              ],
              "ops": [
                {
                  "name": "Trim"
                },
                {
                  "name": "Lowercase"
                }
              ],
              "out": "email"
            },
            {
              "id": "op_name",
              "kind": "map",
              "sources": [
                "full_name"
              ],
              "ops": [
                {
                  "name": "Trim"
                },
                {
                  "name": "Title Case"
                }
              ],
              "out": "fullName"
            },
            {
              "id": "op_src",
              "kind": "literal",
              "value": "nightly-csv",
              "out": "source"
            }
          ],
          "rules": [
            {
              "id": "r_trim",
              "type": "trim",
              "field": "email",
              "severity": "autofix",
              "on": true
            },
            {
              "id": "r_email",
              "type": "validate_email",
              "field": "email",
              "severity": "error",
              "on": true
            }
          ]
        },
        "worker_type": "generic-worker",
        "position": {
          "x": 1020,
          "y": 220
        }
      },
      {
        "id": "upsert",
        "type": "db.upsert",
        "label": "Upsert customers",
        "config": {
          "table": "customers",
          "keys": [
            "email"
          ],
          "row": {
            "email": "steps.process_rows.email",
            "full_name": "steps.process_rows.fullName",
            "source": "steps.process_rows.source"
          }
        },
        "worker_type": "generic-worker",
        "position": {
          "x": 1340,
          "y": 220
        }
      }
    ],
    "edges": [
      {
        "from": "schedule",
        "to": "fetch_csv"
      },
      {
        "from": "fetch_csv",
        "to": "parse_csv"
      },
      {
        "from": "parse_csv",
        "to": "process_rows"
      },
      {
        "from": "process_rows",
        "to": "upsert"
      }
    ]
  }
}

Requirements

The connections and permissions each step needs, from the node contracts in the catalog.

MySQLUsername & password
Used by Upsert customers.
Credential: the password, or {"password": "…"}
- MySQL/MariaDB only in v1.

Setup

Set the CSV url, pick your MySQL connection on Upsert customers, and set table plus the column mapping in row.
Open Normalise columns to edit the operation cards in the Process editor.
To process every row of a large file, insert a For Each over steps.parse_csv.rows before Normalise columns (see the "Fan out list to API" template).

Sample input and output

Synthetic examples: node-contract samples or placeholders, never customer data.

Input

Runs on the cron schedule 0 2 * * * (UTC). There is no incoming payload: the first step fetches the data.

Output

Upsert customers db.upsert

{
  "row_count": 0,
  "table": ""
}

Failure handling

What the graph does when a step fails, read from its retry settings, error branches and error policies.

Download CSV: the request times out after 60 s.
A step that fails, after any retries, fails the run. The failed step is kept with its input and error.