Advanced Transformer use cases

Use the Advanced Transformer to access metadata, state, and utility functions through the context object. The following examples show common implementation patterns.

Add traceability information

Attach processing identifiers to downstream events.

exports.step = function (payload, context) {
  payload.traceId = context.getMetadataValue('traceId');
  payload.sessionId = context.getMetadataValue('sessionId');
  payload.pipelineId = context.getMetadataValue('pipelineId');
  return payload;
};

Example output:

{
  "customerId": "CUST001",
  "usage": 250,
  "traceId": "bc9b4161aea5537424e2932be5e01880|1_b",
  "sessionId": "R-00007391",
  "pipelineId": "3352"
}

Capture source file information

Use source metadata for troubleshooting and audit scenarios.

exports.step = function (payload, context) {
  payload.sourceFile = context.getMetadataValue('fileName');
  payload.sourcePath = context.getMetadataValue('filePath');
  payload.uploadDay = context.getMetadataValue('uploadDay');
  return payload;
};

Example output:

{
  "sourceFile": "usage.csv",
  "sourcePath": "s3://usage-data/usage.csv",
  "uploadDay": "2026-05-29"
}

Include billing context

Expose subscription and catalog information downstream.

exports.step = function (payload, context) {
  payload.chargeModelId = context.getMetadataValue('chargeModelId');
  payload.billingPeriodId = context.getMetadataValue('billingPeriodId');
  return payload;
};

Event deduplication

Use event metadata together with a durable key-value store to ensure an event is processed only once.

exports.step = function (payload, context) {
  const eventId = context.getMetadataValue('eventId');
  const dedupStore = context.getKeyValueState('usage-dedup');
  if (dedupStore.contains(eventId)) {
    return null;
  }
  dedupStore.set(eventId, true);
  return payload;
};

This pattern is commonly used when files may be replayed, sources may resend events, or pipelines need exactly-once business processing.

Create an audit record

Build a complete audit envelope around the original event.

exports.step = function (payload, context) {
  const meta = JSON.parse(context.getMetadataJson());
  return {
    data: payload,
    audit: {
      tenantId: meta.tenantId,
      pipelineId: meta.pipelineId,
      sessionId: meta.sessionId,
      traceId: meta.traceId,
      eventId: meta.eventId
    }
  };
};

Operational logging

Generate troubleshooting logs during processing.

exports.step = function (payload, context) {
  context.log({
    message: "Processing event",
    traceId: context.getMetadataValue('traceId')
  });
  return payload;
};

Logs are emitted only when logging is enabled for the operator.

Working with context state

Context state allows scripts to retain information across records.

State type	Scope	Typical use cases
`context.state`	Current operator key	Running totals, counters
`getKeyValueState()`	Durable operator storage	Deduplication, lookups
`getExternalState()`	Shared Redis-backed state	Cross-pipeline coordination

Running aggregations with context.state

Maintain a running total for each customer.

exports.step = function (payload, context) {
  const total = (context.state.get('totalUsage') || 0) + payload.usage;
  context.state.set('totalUsage', total);
  payload.runningTotalUsage = total;
  return payload;
};

Shared state across pipelines

Use external state when multiple operators or pipelines need access to the same data.

exports.step = function (payload, context) {
  const state = context.getExternalState('global-counters');
  const tenantId = context.getMetadataValue('tenantId');
  const count = (state.get(tenantId) || 0) + 1;
  state.set(tenantId, count);
  payload.tenantEventCount = count;
  return payload;
};

External state requires Redis configuration on the operator.

Bulk lookup operations

Retrieve multiple records in a single call.

exports.step = function (payload, context) {
  const store = context.getKeyValueState('customer-profiles');
  payload.profiles = store.getAll(payload.customerIds);
  return payload;
};

Windowed and accumulator scripts

When processing a collection of events, access metadata by index.

exports.step = function (payloads, context) {
  for (let i = 0; i < payloads.length; i++) {
    payloads[i].traceId = context.getMetadataValueByIndex('traceId', i);
  }
  return payloads;
};

Use this pattern with Advanced Accumulator, window process functions, and scripts that receive event collections.