ETL/Sync Job Builder

Build reliable, incremental data synchronization pipelines.

ETL Job Pattern

// jobs/sync-users.ts

interface

SyncJob

{

name

:

string

;

source

:

"database"

|

"api"

|

"file"

;

destination

:

"database"

|

"warehouse"

|

"s3"

;

schedule

:

string

;

}

export

class

ETLJob

{

constructor

(

private

name

:

string

,

private

watermarkKey

:

string

)

{

}

async

run

(

)

{

console

.

log

(

`

🔄 Starting

${

this

.

name

}

...

`

)

;

try

{

// 1. Get last watermark

const

lastSync

=

await

this

.

getWatermark

(

)

;

console

.

log

(

`

Last sync:

${

lastSync

}

`

)

;

// 2. Extract data

const

data

=

await

this

.

extract

(

lastSync

)

;

console

.

log

(

`

Extracted

${

data

.

length

}

records

`

)

;

// 3. Transform data

const

transformed

=

await

this

.

transform

(

data

)

;

// 4. Load data

await

this

.

load

(

transformed

)

;

// 5. Update watermark

await

this

.

updateWatermark

(

new

Date

(

)

)

;

console

.

log

(

`

✅

${

this

.

name

}

complete

`

)

;

}

catch

(

error

)

{

console

.

error

(

`

❌

${

this

.

name

}

failed:

`

,

error

)

;

throw

error

;

}

}

private

async

extract

(

since

:

Date

)

{

// Extract logic

return

[

]

;

}

private

async

transform

(

data

:

any

[

]

)

{

// Transform logic

return

data

;

}

private

async

load

(

data

:

any

[

]

)

{

// Load logic

}

private

async

getWatermark

(

)

:

Promise

<

Date

>

{

const

watermark

=

await

prisma

.

syncWatermark

.

findUnique

(

{

where

:

{

key

:

this

.

watermarkKey

}

,

}

)

;

return

watermark

?.

lastSync

||

new

Date

(

0

)

;

}

private

async

updateWatermark

(

timestamp

:

Date

)

{

await

prisma

.

syncWatermark

.

upsert

(

{

where

:

{

key

:

this

.

watermarkKey

}

,

create

:

{

key

:

this

.

watermarkKey

,

lastSync

:

timestamp

}

,

update

:

{

lastSync

:

timestamp

}

,

}

)

;

}

}

Watermark Strategy

// Track sync progress

model SyncWatermark {

key String @id

lastSync DateTime

metadata Json?

@@index([lastSync])

}

// Incremental sync using watermark

async

function

syncOrdersIncremental

(

)

{

// Get last sync time

const

watermark

=

await

prisma

.

syncWatermark

.

findUnique

(

{

where

:

{

key

:

"orders_sync"

}

,

}

)

;

const

lastSync

=

watermark

?.

lastSync

||

new

Date

(

0

)

;

// Fetch only new/updated records

const

newOrders

=

await

sourceDb

.

order

.

findMany

(

{

where

:

{

updated_at

:

{

gt

:

lastSync

}

,

}

,

orderBy

:

{

updated_at

:

"asc"

}

,

}

)

;

console

.

log

(

`

📦 Syncing

${

newOrders

.

length

}

orders...

`

)

;

// Process in batches

for

(

let

i

=

0

;

i

<

newOrders

.

length

;

i

+=

100

)

{

const

batch

=

newOrders

.

slice

(

i

,

i

+

100

)

;

await

destinationDb

.

order

.

createMany

(

{

data

:

batch

,

skipDuplicates

:

true

,

// Idempotency

}

)

;

}

// Update watermark to latest record

if

(

newOrders

.

length

>

0

)

{

const

latestTimestamp

=

newOrders

[

newOrders

.

length

-

1

]

.

updated_at

;

await

prisma

.

syncWatermark

.

upsert

(

{

where

:

{

key

:

"orders_sync"

}

,

create

:

{

key

:

"orders_sync"

,

lastSync

:

latestTimestamp

}

,

update

:

{

lastSync

:

latestTimestamp

}

,

}

)

;

}

console

.

log

(

`

✅ Sync complete

`

)

;

}

Idempotent Upsert Pattern

// Idempotent sync - safe to run multiple times

async

function

syncUsersIdempotent

(

users

:

User

[

]

)

{

for

(

const

user

of

users

)

{

await

prisma

.

user

.

upsert

(

{

where

:

{

id

:

user

.

id

}

,

create

:

user

,

update

:

{

email

:

user

.

email

,

name

:

user

.

name

,

updated_at

:

user

.

updated_at

,

}

,

}

)

;

}

}

// Batch upsert for better performance

async

function

syncUsersBatch

(

users

:

User

[

]

)

{

// PostgreSQL: Use ON CONFLICT

await

prisma

.

$executeRaw

`

INSERT INTO users (id, email, name, updated_at)

SELECT * FROM UNNEST(

${

users

.

map

(

(

u

)

=>

u

.

id

)

}

::bigint[],

${

users

.

map

(

(

u

)

=>

u

.

email

)

}

::text[],

${

users

.

map

(

(

u

)

=>

u

.

name

)

}

::text[],

${

users

.

map

(

(

u

)

=>

u

.

updated_at

)

}

::timestamp[]

)

ON CONFLICT (id) DO UPDATE SET

email = EXCLUDED.email,

name = EXCLUDED.name,

updated_at = EXCLUDED.updated_at

WHERE users.updated_at < EXCLUDED.updated_at

`

;

}

Retry Logic with Exponential Backoff

async

function

syncWithRetry

<

T

>

(

operation

:

(

)

=>

Promise

<

T

>

,

maxRetries

:

number

=

3

,

baseDelay

:

number

=

1000

)

:

Promise

<

T

>

{

for

(

let

attempt

=

0

;

attempt

<=

maxRetries

;

attempt

++

)

{

try

{

return

await

operation

(

)

;

}

catch

(

error

)

{

if

(

attempt

===

maxRetries

)

throw

error

;

const

delay

=

baseDelay

*

Math

.

pow

(

2

,

attempt

)

;

console

.

log

(

`

Retry

${

attempt

+

1

}

/

${

maxRetries

}

after

${

delay

}

ms

`

)

;

await

sleep

(

delay

)

;

}

}

throw

new

Error

(

"Max retries exceeded"

)

;

}

// Usage

await

syncWithRetry

(

async

(

)

=>

{

return

await

syncOrders

(

)

;

}

,

3

,

1000

)

;

Change Data Capture (CDC)

// Listen to database changes

import

{

PrismaClient

}

from

"@prisma/client"

;

const

prisma

=

new

PrismaClient

(

)

;

// PostgreSQL: Listen to logical replication

async

function

setupCDC

(

)

{

await

prisma

.

$executeRaw

`

CREATE PUBLICATION orders_publication FOR TABLE orders;

`

;

// Subscribe to changes (using pg library)

const

client

=

await

pg

.

connect

(

)

;

client

.

query

(

"LISTEN orders_changed;"

)

;

client

.

on

(

"notification"

,

async

(

msg

)

=>

{

const

change

=

JSON

.

parse

(

msg

.

payload

)

;

if

(

change

.

operation

===

"INSERT"

||

change

.

operation

===

"UPDATE"

)

{

await

syncOrder

(

change

.

data

)

;

}

}

)

;

}

Conflict Resolution

interface

ConflictResolution

{

strategy

:

"source-wins"

|

"dest-wins"

|

"latest-wins"

|

"merge"

;

}

async

function

syncWithConflictResolution

(

sourceRecord

:

any

,

destRecord

:

any

,

strategy

:

ConflictResolution

[

"strategy"

]

)

{

if

(

strategy

===

"source-wins"

)

{

return

sourceRecord

;

}

if

(

strategy

===

"dest-wins"

)

{

return

destRecord

;

}

if

(

strategy

===

"latest-wins"

)

{

return

sourceRecord

.

updated_at

>

destRecord

.

updated_at

?

sourceRecord

:

destRecord

;

}

if

(

strategy

===

"merge"

)

{

// Merge non-null fields

return

{

...

destRecord

,

...

Object

.

fromEntries

(

Object

.

entries

(

sourceRecord

)

.

filter

(

(

[

_

,

v

]

)

=>

v

!=

null

)

)

,

}

;

}

}

Monitoring & Observability

// Track sync job metrics

interface

SyncMetrics

{

jobName

:

string

;

startTime

:

Date

;

endTime

:

Date

;

recordsProcessed

:

number

;

recordsInserted

:

number

;

recordsUpdated

:

number

;

recordsSkipped

:

number

;

errors

:

number

;

durationMs

:

number

;

}

async

function

logSyncMetrics

(

metrics

:

SyncMetrics

)

{

await

prisma

.

syncMetric

.

create

(

{

data

:

metrics

,

}

)

;

console

.

log

(

`

📊 Sync Metrics

Job:

${

metrics

.

jobName

}

Records:

${

metrics

.

recordsProcessed

}

Inserted:

${

metrics

.

recordsInserted

}

Updated:

${

metrics

.

recordsUpdated

}

Errors:

${

metrics

.

errors

}

Duration:

${

metrics

.

durationMs

}

ms

`

)

;

}

Full ETL Job Example

// jobs/sync-orders-to-warehouse.ts

export

class

OrdersETLJob

extends

ETLJob

{

constructor

(

)

{

super

(

"orders-etl"

,

"orders_warehouse_sync"

)

;

}

async

extract

(

since

:

Date

)

:

Promise

<

Order

[

]

>

{

return

prisma

.

order

.

findMany

(

{

where

:

{

updated_at

:

{

gt

:

since

}

,

}

,

include

:

{

items

:

true

,

user

:

true

,

}

,

orderBy

:

{

updated_at

:

"asc"

}

,

}

)

;

}

async

transform

(

orders

:

Order

[

]

)

:

Promise

<

WarehouseOrder

[

]

>

{

return

orders

.

map

(

(

order

)

=>

(

{

order_id

:

order

.

id

,

user_email

:

order

.

user

.

email

,

total_amount

:

order

.

total

,

item_count

:

order

.

items

.

length

,

status

:

order

.

status

,

order_date

:

order

.

created_at

,

synced_at

:

new

Date

(

)

,

}

)

)

;

}

async

load

(

data

:

WarehouseOrder

[

]

)

:

Promise

<

void

>

{

const

batchSize

=

100

;

for

(

let

i

=

0

;

i

<

data

.

length

;

i

+=

batchSize

)

{

const

batch

=

data

.

slice

(

i

,

i

+

batchSize

)

;

await

warehouseDb

.

$executeRaw

`

INSERT INTO orders_fact (

order_id, user_email, total_amount, item_count,

status, order_date, synced_at

)

VALUES

${

batch

.

map

(

(

o

)

=>

`

(

${

o

.

order_id

}

, '

${

o

.

user_email

}

',

${

o

.

total_amount

}

,

${

o

.

item_count

}

, '

${

o

.

status

}

', '

${

o

.

order_date

}

',

'

${

o

.

synced_at

}

'

)

`

)

.

join

(

","

)

}

ON CONFLICT (order_id) DO UPDATE SET

total_amount = EXCLUDED.total_amount,

status = EXCLUDED.status,

synced_at = EXCLUDED.synced_at

`

;

}

}

}

// Run job

new

OrdersETLJob

(

)

.

run

(

)

;

Scheduling

// Schedule ETL jobs

import

cron

from

"node-cron"

;

// Run every hour

cron

.

schedule

(

"0 * * * *"

,

async

(

)

=>

{

await

new

OrdersETLJob

(

)

.

run

(

)

;

}

)

;

// Run every 15 minutes

cron

.

schedule

(

"/15 * * * "

,

async

(

)

=>

{

await

syncUsersIncremental

(

)

;

}

)

;

// Run nightly at 2 AM

cron

.

schedule

(

"0 2 * * *"

,

async

(

)

=>

{

await

fullDataSync

(

)

;

}

)

;

Error Handling & Recovery

async

function

syncWithErrorHandling

(

)

{

const

checkpoint

=

await

getCheckpoint

(

)

;

let

processedRecords

=

0

;

try

{

const

records

=

await

fetchRecords

(

checkpoint

)

;

for

(

const

record

of

records

)

{

try

{

await

processRecord

(

record

)

;

processedRecords

++

;

// Save checkpoint every 100 records

if

(

processedRecords

%

100

===

0

)

{

await

saveCheckpoint

(

record

.

id

)

;

}

}

catch

(

error

)

{

// Log error but continue

console

.

error

(

`

Failed to process record

${

record

.

id

}

:

`

,

error

)

;

await

logFailedRecord

(

record

.

id

,

error

)

;

}

}

await

saveCheckpoint

(

"completed"

)

;

}

catch

(

error

)

{

// Critical failure - job will retry from checkpoint

console

.

error

(

"Job failed:"

,

error

)

;

throw

error

;

}

}

Best Practices

Incremental sync

Use watermarks, don't full-scan

Idempotent operations

Safe to retry

Batch processing

Process 100-1000 records at a time

Checkpointing

Resume from failure point

Retry with backoff

Handle transient failures

Monitor metrics

Track job health

Test thoroughly

Including failure scenarios Output Checklist ETL job class created Watermark tracking implemented Incremental sync logic Idempotent upsert operations Retry logic with backoff Conflict resolution strategy Monitoring and metrics Error handling and recovery Job scheduling configured Testing including failure cases