Your first Pipeline¶

One Pipe collects readings. Another converts them. A Pipeline runs those steps in the right order and gives the workflow a name.

Continue with the two readings from Incremental sync.

flowchart LR
    A[(readings)] --> B[temperatures]
    B --> C[fahrenheit]

Add the transformation¶

Run this Python snippet in the same workspace:

import meerschaum as mrsm

fahrenheit = mrsm.Pipe(
    'sql:local', 'fahrenheit', instance='sql:local',
    columns={'datetime': 'ts', 'primary': 'id'},
    dtypes={'ts': 'datetime'},
    parameters={'sql': """
        SELECT id, ts,
               ROUND(temperature * 9.0 / 5 + 32, 1) AS fahrenheit
        FROM {{ Pipe('sql:local', 'temperatures', instance='sql:local') }}
    """},
)
fahrenheit.register()

The {{ Pipe(...) }} reference finds the parent's table and tells Meerschaum that temperatures comes first.

Save, preview, run¶

mrsm register pipelines climate -i sql:local -c sql:local -m fahrenheit --parents
mrsm plan pipelines climate -i sql:local
mrsm sync pipelines climate -i sql:local
mrsm show data -i sql:local -m fahrenheit

plan shows the work without moving data. sync runs the Pipeline and records its results. You should see 70.7°F and 71.6°F.

import meerschaum as mrsm

pipeline = mrsm.Pipeline('climate', instance='sql:local')
print(pipeline.plan())
run = pipeline.sync()
print(run.success, run.inserted, run.updated)

Let the graph do the remembering¶

When you add a reading and run climate again, Meerschaum syncs the parent first. The child then uses the data available from its parent. Independent branches can run in parallel.

A saved Pipeline remembers its members and settings. A job supplies a schedule. Deleting the Pipeline's saved definition keeps its Pipes and their data.

See it in the console →

For backfills, retries, and larger graphs, continue to the Pipeline reference.