---
title: "Processor types for custom dataflow integrations"
slug: "processor-types-for-custom-dataflow-integrations"
updated: 2026-08-25T11:09:29Z
published: 2026-08-25T11:09:29Z
canonical: "support.lusid.com/processor-types-for-custom-dataflow-integrations"
---

> ## Documentation Index
> Fetch the complete documentation index at: https://support.lusid.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Processor types for custom dataflow integrations

This reference article describes the processor types available in [the **Dataflow editor**](/v1/docs/configuration-driven-integrations).

Each processor is a single step in a pipeline. For an overview of how processors fit together in pipelines, see [How custom dataflow integrations work](/v1/docs/configuration-driven-integrations).

> [!CAUTION]
> Important
> 
> This feature is in [Early Access](/v1/docs/what-is-the-lusid-feature-release-lifecycle#early-access) and is subject to change.

## Common settings

Every processor shares the following settings, regardless of type:

| Setting | Description |
| --- | --- |
| **Output mode** | What happens to records as they leave this processor. See [Output modes](/v1/docs/configuration-driven-integrations#output-modes). |
| **Input record type** | The output record types this processor takes in records from |
| **Output record type** | Records flow out of processors as this named record type |
| **Batch size** | The maximum number of records the output flow can hold before upstream processors pause |

## Ingest

The **Ingest** processor reads data from a source, parses it into records, and passes those records to the next processor. LUSID creates a lineage record for each file it processes.

**Ingest** is always the first processor in a pipeline.

| Setting | Description |
| --- | --- |
| **Source type** | Where to read data from: `Drive`, `Sftp`, `Api`, `S3`, `ActiveMQ`, `MongoDB`, `Lusid`, `Postgres`, `Snowflake` |
| **Parser type** | The format of the source data: `Csv`, `Json`, `Xml`, `FixedWidth`, `Excel`, `Fix`, or `SwiftMt` |
| **Path / pattern** and **Root path** | The file path, URL, queue name, or collection (depending on the source type) to read from |
| **Boundary field** | The field on upstream records whose value change triggers a new call to the data source. When specified, the processor buffers and sorts upstream records by the boundary field. |

For example, you could include the following segments in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

**CSV in LUSID Drive**

```json
{
  "ProcessorType": "Ingest",
  "Sequence": 1,
  "OutputFlows": { "default": "ingested" },
  "Config": {
    "sourceType": "Drive",
    "parserType": "Csv",
    "path": "/integrations/jpm-imos/positions/*.csv",
    "delimiter": "|",
    "recordType": "ValuedHolding",
    "batchSize": 500
  }
}
```

**API call**

```json
{
  "ProcessorType": "Ingest",
  "Sequence": 2,
  "InputFlows": { "default": "portfolios" },
  "OutputFlows": { "default": "positions" },
  "Config": {
    "sourceType": "Api",
    "parserType": "Json",
    "path": "https://api.custodian.com/v1/positions",
    "auth": { "type": "bearer", "token": "SECRET:custodian-token" },
    "boundaryField": "AccountId",
    "inputDerivedQueryParams": [
      { "paramName": "account", "sourceField": "AccountId" }
    ]
  }
}
```

## Map transform

The **Map transform** processor reshapes records:

- Renames fields
- Applies string transforms
- Tags records with a type
- Expands a single record into multiple records

### Field mappings

Rules for renaming or transforming fields. Each rule specifies a source field, target field, and an optional transform function.

| Setting | Description |
| --- | --- |
| **Source field** | The source file column you want to apply the transformation to |
| **Target field** | A field name for the transformed data |
| **Transform** | **None** | Simply renames the field without transforming any data |
| **Uppercase** | Sets all field values to UPPERCASE |
| **Lowercase** | Sets all field values to lowercase |
| **Trim** | Removes surrounding whitespace from a field value, for example `“  AAPL  “` → `“AAPL“` |
| **Strip from plus** | Removes everything from the first `+` character onwards |

### Value maps

A list of key-value pairs for mapping specific values, which you can then reference in your [value sources](/v1/docs/processor-types-for-custom-dataflow-integrations#value-sources). Each list item should contain:

- A unique reference **Map name**
- A list of key-value pairs

For example, you could import the following mapping JSON to automatically populate the value maps settings:

```json
{
  "valueMaps": [
    {
      "map": {
        "BUY": "Buy",
        "SELL": "Sell",
        "DIVIDEND": "StockDividend"
      },
      "name": "TransactionTypeMap"
    },
    {
      "map": {
        "Monthly": "1M",
        "Quarterly": "3M",
        "Semi-Annual": "6M",
        "Annual": "1Y"
      },
      "name": "FrequencyMap"
    }
  ]
}
```

![](https://cdn.document360.io/d575ad81-c0ed-4980-bbd1-d59ac5c3de82/Images/Documentation/image-1Z7NK20M.png?sv=2026-02-06&spr=https&st=2026-11-09T07%3A08%3A18Z&se=2026-11-09T08%3A08%3A18Z&sr=c&sp=r&sig=Z%2F0NnXsuPz65azscW%2BUeAMRFPzkmM1Xhw3gjJ308LSs%3D)

### Value sources

Rules for deriving new fields; supports taking the first non-empty value from a list, concatenating fields, setting default values, and mathematical expressions.

| Setting | Description |
| --- | --- |
| **Name** | A name to use when referring to this value source |
| **Source kind** | **Column** | One or more columns from the source file that the value should be sourced from. Specifying a [value conversion](/v1/docs/processor-types-for-custom-dataflow-integrations#value-conversion) tells the processor how to combine multiple columns. |
| **Conditional** | One or more rules containing complex logic for value transformations based on conditions. See [conditional value sources](/v1/docs/processor-types-for-custom-dataflow-integrations#constant-value). |
| **Constant** | Hardcode a specific value |
| **Expression** | A derivation formula that instructs the processor how to automatically calculate values from one or more fields. See [supported operations](/v1/docs/derivation-formulas-part-1-operations-and-allowed-values). |
| **MathematicalExpression** | A formula that can use other value sources to output a value. See [mathematical expression](/v1/docs/processor-types-for-custom-dataflow-integrations#mathematical-expression). |
| **ValueMapLookup** | The name of a [value map](/v1/docs/processor-types-for-custom-dataflow-integrations#value-maps) to use for the value source |

> [!NOTE]
> Best practice
> 
> - [**Reuse value sources**](/v1/docs/record-capture-framework#reuse-valuesources): Define value sources once and reference them multiple times.
> - [Use value maps instead of multiple conditional checks.](/v1/docs/record-capture-framework#use-valuemaps-for-transformations)
> - Set default values for the processor to fallback on when it encounters unexpected values.

#### Reference: Date formats

One or more of the following date formats; the processor attempts to apply each format to the value in order until it’s successful:

- `yyyy-MM-dd` (recommended)
- `yyyy-MM-ddTHH:mm:ssZ` (recommended)
- `dd/MM/yyyy`
- `MM/dd/yyyy`
- `yyyy-MM-ddTHH:mm:ss`
- `dd-MMM-yyyy`
- `yyyyMMdd`

#### Reference: Conditional value sources

Each rule should contain:

- **Condition**: A [LUSID filtering syntax expression](/v1/docs/filtering-information-retrieved-from-lusid) to evaluate the value source against. It must use the format `<Value source> <Operator> <Specific value/source>`.

> [!NOTE]
> Supported operators
> 
> The integration framework supports the following operators in addition to the standard LUSID filtering syntax [supported operators](/v1/docs/filtering-information-retrieved-from-lusid#appendix-a-supported-operators):
> 
> - `InMapKeys`: Value is a **Key** in the defined [value maps](/v1/docs/processor-types-for-custom-dataflow-integrations#value-maps), for example `TypeValue InMapKeys ValidTypesMap`
> - `InMapValues`: Value is a **Value** in the defined [value maps](/v1/docs/processor-types-for-custom-dataflow-integrations#value-maps), for example `StatusValue InMapValues StatusMap`
- **Result**: The value to return if the condition is met; this can be any of the following:
  - A specific value
  - One of the other defined value sources

![](https://cdn.document360.io/d575ad81-c0ed-4980-bbd1-d59ac5c3de82/Images/Documentation/image-LP7U1D2Z.png?sv=2026-02-06&spr=https&st=2026-11-09T07%3A08%3A18Z&se=2026-11-09T08%3A08%3A18Z&sr=c&sp=r&sig=Z%2F0NnXsuPz65azscW%2BUeAMRFPzkmM1Xhw3gjJ308LSs%3D)

#### Reference: Column references

A list of columns from the source file that the value should be sourced from. Specifying a [value conversion](/v1/docs/processor-types-for-custom-dataflow-integrations#value-conversion) tells the processor how to combine multiple columns.

#### Reference: Value conversion

Specify how to combine multiple [column references](/v1/docs/processor-types-for-custom-dataflow-integrations#column-references1):

- `None`: Takes the first **Column references** value
- `Concatenate`: Joins multiple **Column references** values together using the specified [**Separator**](/v1/docs/processor-types-for-custom-dataflow-integrations#separator)
- `FirstNonEmpty`: Uses the first non-empty **Column references** value
- `RegexExtraction`: The processor applies a regex pattern to the value from either:

> [!NOTE]
> Note
> 
> - Regex patterns time out after 1 second; to avoid timeouts, use patterns that target specific locations (for example, `^ISIN:\\s*([A-Z0-9]{12})` rather than `.*ISIN.*`).
> - The framework only extracts the first parenthesised group in your pattern (if used).
  - The first column in **Column references**
  - If no **Column references** are specified, the first item listed in [Value sources](/v1/docs/processor-types-for-custom-dataflow-integrations#value-sources)

#### Reference: Currency identifier

Check the box to add the prefix `CCY_` to each value, for example `USD` → `CCY_USD`.

#### Reference: Separator

Defaults to a space (`“ “`).

The separator to use when [Value conversion](/v1/docs/processor-types-for-custom-dataflow-integrations#value-conversion) is set to `Concatenate`. For example, specifying `_` as separator transforms the values `FUND-001`, `US0378331005`, `2025-03-10` → `FUND-001_US0378331005_2025-03-10`.

#### Reference: Mathematical expression

Specify a formula that can use other value sources to output a value. You can use any of the following in your formula:

- Names of `ValueSources`
- Numeric literals
- `+`, `-`, `*`, `/`

> [!NOTE]
> Note
> 
> The framework evaluates strictly left-to-right - there is no operator precedence.

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "MapTransform",
  "Sequence": 2,
  "InputFlows": { "default": "ingested" },
  "OutputFlows": { "default": "mapped" },
  "Config": {
    "mappings": [
      { "source": "AccountNum", "target": "PortfolioCode", "transform": "TRIM" },
      { "source": "TickerSymbol", "target": "Ticker", "transform": "UPPERCASE" }
    ],
    "recordTypeMappings": [
      { "recordType": "Cash", "condition": "InstrumentType == 'CASH'" },
      { "recordType": "Filtered", "condition": "Quantity == '0'" }
    ],
    "recordExpansions": [
      {
        "when": "InstrumentType == 'FxForward'",
        "tagField": "_Leg",
        "emit": ["BuyLeg", "SellLeg"]
      }
    ],
    "ValueSources": [
      {
        "name": "Side",
        "mathematicalExpression": "Units * 1",
        "defaultValue": "Buy"
      }
    ]
  }
}
```

## Instrument resolver

The **Instrument resolver** processor looks up each record’s instrument in LUSID and adds the resolved [LUID](/v1/docs/what-is-a-lusid-instrument-id-or-luid) to the record.

Records flow out of the processor in one of two output flows:

- One output flow for successfully resolved records
- One output flow for records that could not be resolved

| Setting | Description |
| --- | --- |
| **Identifier mode** | Configure the processor to resolve against a single identifier type, or a prioritised fallback hierarchy - the processor uses the first identifier that resolves uniquely. |
| **Identifier type** | The [LUSID instrument identifier type](/v1/docs/what-are-unique-and-non-unique-identifiers) to look up for each record. |
| **Identifier source** | The field on the record containing the identifier value to look up in LUSID. |
| **Resolved LUID field** | A name for the LUID field to attach to resolved records |
| **Resolved identifier type field** | In `Hierarchy` mode, a name for the field that records which identifier type matched |
| **Resolved output flow** | The name of the output flow for successfully resolved records |
| **Unresolved output flow** | The name of the output flow for records that could not be resolved |
| **Stamp fields** | Additional LUSID instrument fields to read from LUSID and attach to resolved records |
| **Search scope** | The LUSID scope to look up instruments in |
| **Sort key** | The processor buffers and sorts records by this record field before upserting |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "InstrumentResolver",
  "Sequence": 3,
  "InputFlows": { "default": "mapped" },
  "OutputFlows": { "resolved": "resolved", "unresolved": "to_create" },
  "Config": {
    "identifierHierarchy": [
      { "type": "Isin",     "source": "ISIN" },
      { "type": "Cusip",   "source": "CUSIP" },
      { "type": "Currency","source": "Currency" }
    ],
    "stampFields": ["InstrumentType", "Name"],
    "searchScope": "default",
    "batchSize": 200
  }
}
```

## Portfolio resolver

The **Portfolio resolver** processor looks up each record’s portfolio in LUSID and adds the resolved [portfolio scope and code](/v1/docs/portfolios) to the record.

Records flow out of the processor in one of two output flows:

- One output flow for successfully resolved records
- One output flow for records that could not be resolved

| Setting | Description |
| --- | --- |
| **Resolution mode** | Configure the processor to resolve against portfolios filtered by properties and a linked code, a portfolio group, or using each record’s portfolio code value. [See resolution modes below.](/v1/docs/processor-types-for-custom-dataflow-integrations#linked-property-resolution-mode) |
| **Portfolio code source** | The field on the record containing the portfolio code value to look up in LUSID |
| **Resolved scope field** | A name for the resolved portfolio scope field to attach to resolved records |
| **Resolved code field** | A name for the resolved portfolio code field to attach to resolved records |
| **Resolved output flow** | The name of the output flow for successfully resolved records |
| **Unresolved output flow** | The name of the output flow for records that could not be resolved |

### Resolution modes

- **Linked property**: Fetches portfolios filtered by required properties, then matches records using a linked code property. Use this option when your portfolios have a property that maps to the source system’s account identifier.
- **Portfolio group**: Resolves against portfolios in a specified portfolio group.
- **Direct**: Resolves each record using its portfolio code value and a specified scope. Use this option when your records already contain the LUSID scope and code.

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "PortfolioResolver",
  "Sequence": 4,
  "InputFlows": { "default": "resolved" },
  "OutputFlows": { "resolved": "ready", "unresolved": "skipped" },
  "Config": {
    "resolutionMode": "LinkedProperty",
    "portfolioCodeSource": "AccountNumber",
    "linkedCodeProperty": "Portfolio/Finbourne-Examples/CustodianAccount",
    "requiredProperties": [
      { "key": "Portfolio/Finbourne-Examples/Custodian", "value": "Acme" }
    ]
  }
}
```

## Enrichment

The **Enrichment** processor adds extra fields to each record by looking up a value against mappings you specify in the processor configuration or an external API.

| Setting | Description |
| --- | --- |
| **Enrichment source** | Where to look up enrichment data: `StaticMap` or `Api` |
| **Enrichment key field** | The name of a record field - the value of this record field is used as the lookup key |
| **On no match** | What to do when no field match is found: - **Ignore**: Pass the record through the processor unchanged - **Default**: Apply default values to the record - **Fail**: Stop the pipeline |
| **Default values** | Field values to apply when **On no match** is set to `Default` |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

**Static map**

```json
{
  "ProcessorType": "Enrichment",
  "Config": {
    "source": "StaticMap",
    "enrichmentKeyField": "Currency",
    "map": {
      "USD": { "Region": "US", "G10": "true" },
      "EUR": { "Region": "EMEA", "G10": "true" }
    },
    "fields": [
      { "sourceField": "Region", "targetField": "Currency_Region" },
      { "sourceField": "G10", "targetField": "Currency_IsG10" }
    ],
    "onNoMatch": "default",
    "defaultValues": { "Currency_Region": "Other", "Currency_IsG10": "false" }
  }
}
```

**API**

```json
{
  "ProcessorType": "Enrichment",
  "Config": {
    "source": "LusidHoldings",
    "fanOut": true,
    "enrichmentKeyField": "PortfolioId",
    "scopeField": "PortfolioScope",
    "codeField": "PortfolioCode",
    "effectiveAt": "{{DATE:T}}",
    "carryFields": ["PortfolioId"],
    "onNoMatch": "ignore"
  }
}
```

## Merge

The **Merge** processor combines multiple input flows into a single output flow.

| Setting | Description |
| --- | --- |
| **Strategy** | How to combine the input flows: - **Concat**: Passes records through as they arrive from an input - **GroupByKey**: Groups incoming records by the specified **Merge key field**, then outputs the records in order |
| **Merge key field** | The name of the record field to sort by when **Strategy** is set to `GroupByKey` |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "Merge",
  "InputFlows": { "positions": "positions_raw", "cash": "cash_raw" },
  "OutputFlows": { "default": "merged_raw" },
  "Config": {
    "strategy": "GroupByKey",
    "mergeField": "PortfolioCode"
  }
}
```

## Query

The **Query** processor runs a SQL query against a [Luminesce](/v1/docs/luminesce-and-lumipy), Postgres or Snowflake database and emits each result row as a record. You can run this as either:

- A standalone query with no input flows; or,
- A dynamic query, where the processor receives records from an output flow and uses the field values from those records as query parameters

| Setting | Description |
| --- | --- |
| **Database type** | `Luminesce`, `Postgres`, or `Snowflake` |
| **Connection string** | For non-Luminesce database types: the database connection string; supports `SECRET:` tokens for credentials |
| **SQL query** | The parameterised SQL query to run |
| **Query name** | A name to identify your query in logs |
| **Query timeout** | How long to wait before the query times out |
| **Static parameters** | One or more static key and value pairs to apply on every execution of the query |
| **Dynamic parameters** | One or more parameters sourced from your upstream records; note the query runs once per batch of records |
| **Max rows** | A safety limit on the number of rows the query can return; you can set the value to `-1` to disable the limit |
| **On empty input** | What to do when no value exists for a dynamic parameter: - **Skip**: Log the skipped record and complete with no output - **Error**: Stop the pipeline |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "Query",
  "Sequence": 4,
  "InputFlows": { "default": "trades" },
  "OutputFlows": { "default": "instrument_meta" },
  "Config": {
    "databaseType": "Postgres",
    "connectionString": "Host=ref.Finbourne-Examples.com;Database=ref;SslMode=Require",
    "credentialsKey": "ref-pg",
    "query": "SELECT isin, sector, country FROM reference.instruments WHERE isin = ANY(@isins)",
    "dynamicParameters": [
      { "paramName": "isins", "sourceField": "ISIN", "distinct": true, "batchSize": 500 }
    ],
    "recordType": "InstrumentMeta"
  }
}
```

## Deduplication

The **Deduplication** processor groups records by a particular field and passes on only one record per group, either:

- The record with the largest **Order by field** value (`latest`, the default); or
- The record with the smallest **Order by field** value (`earliest`)

| Setting | Description |
| --- | --- |
| **Group by field** | The name of the record field to group by |
| **Order by field** | The name of the record field used to determine which record should be kept for each group |
| **Keep** | How the processor determines which record to keep per group: - `latest`: Keeps the largest **Order by** field value (typically the most recent date) - `earliest`: Keeps the smallest **Order by field** value (typically the oldest date) |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "Dedup",
  "InputFlows": { "default": "batchruns" },
  "OutputFlows": { "default": "latest_per_batch" },
  "Config": {
    "groupByField": "batch.id",
    "orderByField": "asOf",
    "keep": "latest"
  }
}
```

## Rename

The **Rename** processor applies a list of renaming rules to the file name on each record flowing through a file-transfer pipeline. For example, you might rename files after an SFTP download or decryption step but before an upload.

Every rule that matches the filter condition runs in sequence; the renamed output of one rule becomes the input for the next rule.

> [!NOTE]
> Note
> 
> Rules with no filter condition are always applied.

| Setting | Description |
| --- | --- |
| **Rename rules** | **Filter expression** | A [LUSID filtering syntax](/v1/docs/filtering-information-retrieved-from-lusid) condition that must be true for the rule to apply. If left blank, the rule always applies. |
| **Derive expression** | A LUSID derivation formula that determines the new file name. Supports [concatenation](/v1/docs/processor-types-for-custom-dataflow-integrations#value-conversion), [conditionals](/v1/docs/processor-types-for-custom-dataflow-integrations#constant-value), and [date/number formatting](/v1/docs/processor-types-for-custom-dataflow-integrations#reference-date-formats). |
| **On no match** | What to do when none of the filters match: - `passthrough`: Pass on the record with its original file name - `error`: Mark and log the record as failed |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "Rename",
  "Config": {
    "rules": [
      {
        "filter": "FileName like 'prices_%.csv'",
        "derive": "'prices_' + Concat(BusinessDate as date format 'yyyyMMdd') + '.csv'"
      }
    ],
    "onNoMatch": "passthrough"
  }
}
```

## Data quality check

The **Data quality check** processor runs data quality rules you configure against each record. You can choose what happens when a rule fails:

- Flag the record and continue
- Stop the pipeline
- Attach the errors to the record for downstream handling

| Setting | Description |
| --- | --- |
| **Check type** | The type of data quality validation to run |
| **Field** | The record field this rule validates |
| **Fail mode** | What to do when a rule fails: - **Flag**: Marks the record as filtered, log a warning, and pass it downstream - **Reject**: Stop the pipeline - **Report**: Attach error details to the record in a `_DQErrors` field and pass the record downstream |
| **Error message** | A message to log if the rule fails |
| **Max value** and **Min value** | Used for `Range` checks; lower and upper bounds for the value |
| **Allowed values** | Used for `InSet` checks; enter one or more allowed values |
| **Regex pattern** | Used for `Regex` checks |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "DQCheck",
  "Sequence": 5,
  "Config": {
    "failMode": "flag",
    "rules": [
      { "name": "isin-required", "check": "Required",    "field": "ISIN",      "when": "InstrumentType != 'CASH'" },
      { "name": "isin-format",   "check": "Regex",       "field": "ISIN",      "pattern": "^[A-Z]{2}[A-Z0-9]{9}\\d$" },
      { "name": "qty-positive",  "check": "GreaterThan", "field": "Quantity",  "value": "0", "severity": "warning" },
      { "name": "ccy-allowed",   "check": "In",          "field": "Currency",  "values": ["USD","EUR","GBP","JPY"] },
      { "name": "trade-unique",  "check": "UniqueInBatch","field": "TradeId" },
      { "name": "trade-date",    "check": "DateFormat",  "field": "TradeDate", "format": "yyyyMMdd" }
    ]
  }
}
```

## File export

The **File export** processor turns records into a file and writes the file to your specified destination. You can export files with the following formats:

- CSV
- JSON
- XML
- Fixed-width

| Setting | Description |
| --- | --- |
| **Output format** | The format of the output file: `Csv`, `Json`, `Xml`, or `FixedWidth` |
| **Include headers** | When enabled and **Output format** is CSV, includes a header row in the file |
| **Delimiter** | Used for CSV output files; the column delimiter |
| **Pretty print** | When enabled, formats the JSON output file with indentation and line breaks for readability |
| **Column order** | Used for CSV and fixed-width output files; an ordered list of field names to explicitly order the columns in the output file |
| **Destination type** and **Output path** | Where to write the output file: `Drive`, `S3`, `Sftp`, or `Local` |
| **Transform pipeline** | Optionally compress the output before writing it to the destination: `Gzip`, `Zip`, `Pgp` |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after exporting them as a file |
| **Record type filter** | If specified, only process records with this record type attached |
| **Pass through unmatched** | Passes records that do not match the **Record type filter** downstream unchanged (rather than dropping them) |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "SinkFile",
  "Config": {
    "format": "Csv",
    "columnOrder": ["PortfolioCode", "Ticker", "Quantity"],
    "includeHeaders": true,
    "pipeline": [
      { "step": "compress", "type": "GZip" },
      { "step": "encrypt", "type": "Pgp", "publicKeyRef": "{{SECRET:client-pgp-pub}}" }
    ],
    "destination": {
      "type": "Sftp",
      "host": "sftp.example.com",
      "port": 22,
      "credentialsKey": "sftp-creds",
      "remotePath": "/outbound/positions-{{COBDATE:yyyyMMdd}}.csv"
    },
    "checksumFile": { "remotePath": "/outbound/positions.csv.sha256", "algorithm": "sha256" },
    "continueAfterSink": false
  }
}
```

## Transfer

The **Transfer** processor moves files from one location to another without parsing them. This can be useful for delivering files to an external system or encrypting files in transit.

LUSID keeps a log of each file the processor transfers.

| Setting | Description |
| --- | --- |
| **Source type** and **Source path** | Where to source the file from: `Drive`, `Sftp`, `S3`, or `Local` |
| **Delete after transfer** | When enabled, the processor deletes the file in the source location after it has been successfully transferred to the destination. If the processor is unable to delete the source file after transfer, it logs the failure as a warning. |
| **Destination type** and **Destination path** | Where to move the file to: `Drive`, `Sftp`, `S3`, or `Local` |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "Transfer",
  "Config": {
    "source": {
      "type": "Drive",
      "path": "/integrations/outbound/*.csv",
      "deleteAfterTransfer": false
    },
    "destination": {
      "type": "Sftp",
      "path": "/incoming/",
      "host": "sftp.example.com",
      "pipeline": [{ "step": "encrypt", "type": "Pgp", "publicKeyRef": "client-pgp-pub" }]
    }
  }
}
```

## Upsert corporate actions

*More information coming soon.*

## Upsert custom entities

The **Upsert custom entities** processor creates or updates [custom entities](/v1/docs/extending-lusids-data-model-using-custom-entities) in LUSID from records. If the custom entity type does not yet exist in your domain, LUSID creates it for you on the first run.

| Setting | Description |
| --- | --- |
| **Entity type name** | The custom entity type name |
| **Display name** and **Description** | A friendly display name and description for the custom entity type |
| **Create type if not exists** | When enabled, creates the [custom entity type definition](/v1/docs/extending-lusids-data-model-using-custom-entities#defining-a-custom-entity-type) automatically on the first run |
| **Identifiers** | At least one identifier for the custom entity, comprised of a static `Type`, `Scope`, and a value `Source field`; see [Understanding user-defined identifiers](/v1/docs/extending-lusids-data-model-using-custom-entities#understanding-userdefined-identifiers) |
| **Field schema** | The definitions for each field on the custom entity type; see [Defining a custom entity type](/v1/docs/extending-lusids-data-model-using-custom-entities#defining-a-custom-entity-type) |
| **Field mappings** | Where to source the values for each custom entity field |
| **Property mappings** | LUSID [properties](/v1/docs/properties) to decorate onto custom entities with a value from your specified source field |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as custom entities in LUSID |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertCustomEntities",
  "Config": {
    "entityTypeName": "FundManager",
    "createTypeIfNotExists": true,
    "displayNameSource": "ManagerName",
    "descriptionSource": "ManagerDescription",
    "fieldSchema": [
      { "Name": "Region", "Type": "string", "Required": true, "Lifetime": "Perpetual" },
      { "Name": "Aum", "Type": "decimal", "Required": false, "Lifetime": "TimeVariant" }
    ],
    "fieldMappings": [{ "FieldName": "Aum", "SourceField": "AssetsUnderManagement" }],
    "identifiers": [{ "Scope": "default", "Type": "ManagerCode", "SourceField": "Code", "Required": true }],
    "propertyMappings": [
      { "PropertyKey": "CustomEntity/default/Tier", "SourceField": "Tier", "DataType": "string", "Unit": null }
    ],
    "batchSize": 100
  }
}
```

## Upsert holdings

The **Upsert holdings** processor creates or updates [holdings](/v1/docs/holdings) in LUSID from records. You can process holdings in two ways:

- **Adjust holdings**: Add to the existing portfolio holdings
- **Set holdings**: Replace the existing portfolio holdings

| Setting | Description |
| --- | --- |
| **Mode** | How the holdings you upsert should behave with your existing holdings in LUSID: either `Adjust` or `Set`. [Read more on adjusting or setting holdings.](/v1/docs/how-do-i-manually-adjust-or-set-holdings) |
| **Portfolio scope** | The LUSID portfolio scope to write holdings into |
| **Portfolio code source** | The record field to source the portfolio code from |
| **EffectiveAt source** | The record field to source the effectiveAt datetime from |
| **EffectiveAt format** | If the effectiveAt value is not ISO 8601-formatted, specify a date format (as listed in [date formats](/v1/docs/processor-types-for-custom-dataflow-integrations#date-formats)) |
| **Instrument identifier type** | The three-stage [LUSID instrument identifier key](/v1/docs/understanding-instrument-identifiers), for example `Instrument/default/LusidInstrumentId` |
| **<Holdings field> source** | The record field to populate each holding value from; [read more on LUSID’s holdings fields](/v1/docs/how-do-i-manually-adjust-or-set-holdings) |
| **Property mappings** | Each property mapping configures the processor to decorate a LUSID property onto holdings with a value from your specified source field |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as holdings in LUSID |
| **Aggregate duplicates by instrument** | When enabled in `Set` mode, the processor sums the units for duplicate instrument records to form a single holding |
| **Sort key** | In `Set` mode, the processor buffers and sorts records by this record field before upserting |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertHoldings",
  "InputFlows": { "default": "ready" },
  "Config": {
    "mode": "Set",
    "scope": "MyPortfolioScope",
    "portfolioCodeSource": "_ResolvedPortfolioCode",
    "instrumentIdentifierType": "Instrument/default/LusidInstrumentId",
    "instrumentIdentifierSource": "_ResolvedLuid",
    "unitsSource": "Quantity",
    "currencySource": "Ccy",
    "costAmountSource": "BookCost",
    "costCurrencySource": "Ccy",
    "effectiveAtSource": "PositionDate",
    "effectiveAtFormat": "yyyy-MM-dd",
    "dateTimeZone": "America/Toronto",
    "subHoldingKeys": ["Strategy"],
    "portfolioBaseCurrency": "CAD",
    "aggregateDuplicatesByInstrument": true,
    "propertyMappings": [
      { "propertyKey": "Holding/Finbourne-Examples/Custodian", "sourceField": "Custodian", "dataType": "string" }
    ],
    "custodianAccountScope": "Custodians",
    "custodianAccountCodeSource": "AccountNumber",
    "batchSize": 100,
    "sortKey": "_ResolvedPortfolioCode"
  }
}
```

## Upsert instrument

The **Upsert instrument** processor creates or updates [instrument definitions in LUSID](/v1/docs/instruments) from records. You can assign instrument types by condition, with a fallback type for records that don’t match a condition.

| Setting | Description |
| --- | --- |
| **Identifier mapping source field** | The record field or value source for the identifier |
| **LUSID identifier type** | The [LUSID instrument identifier type](/v1/docs/understanding-instrument-identifiers) to populate with a value from the source field |
| **Name source field** | The record field to use as the instrument name |
| **Default instrument type** | A fallback instrument type for the processor to assign when no type condition matches |
| **Domestic currency source** | The record field to use as the instrument’s domestic currency |
| **Property mappings** | Each property mapping configures the processor to decorate a LUSID property onto instruments with a value from your specified source field |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as instruments in LUSID |
| **Resolved LUID field** | When **Continue after sink** is enabled, uses this name for the field containing the record’s resolved [LUID](/v1/docs/what-is-a-lusid-instrument-id-or-luid) |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertInstrument",
  "Config": {
    "nameSource": "InstrumentName",
    "instrumentTypeSource": "AssetType",
    "instrumentTypeMap": { "BOND": "Bond", "EQ": "Equity" },
    "defaultInstrumentType": "SimpleInstrument",
    "identifierType": "Isin",
    "identifierMappings": [
      { "lusidIdentifierType": "Isin", "sourceField": "ISIN", "isRequired": true },
      { "lusidIdentifierType": "Cusip", "sourceField": "CUSIP", "isRequired": false }
    ],
    "propertyMappings": [{ "propertyKey": "Instrument/default/Sector", "sourceField": "Sector", "dataType": "string" }],
    "instrumentTypeConfigs": [
      { "instrumentType": "Bond", "maturityDateSource": "Maturity", "couponRateSource": "Coupon" }
    ],
    "batchSize": 100,
    "continueAfterSink": true,
    "resolvedLuidField": "_ResolvedLuid"
  }
}
```

## Upsert legal entities

The **Upsert legal entities** processor creates or updates [legal entities](/v1/docs/representing-institutions-in-lusid-using-legal-entities) in LUSID from records. Before upserting, the processor deduplicates records within a batch by their `idTypeScope`.

| Setting | Description |
| --- | --- |
| **Identifier scope** | The `idTypeScope` of the legal entity (the second stage in the 3-stage key); see [Understanding legal entity identifiers](/v1/docs/representing-institutions-in-lusid-using-legal-entities#understanding-userdefined-identifiers) |
| **Identifier type** | The `idTypeCode` of the legal entity (the third stage in the 3-stage key); see [Understanding legal entity identifiers](/v1/docs/representing-institutions-in-lusid-using-legal-entities#understanding-userdefined-identifiers) |
| **Identifier value source** | The record field to use as the legal entity’s `code` value; see [Understanding legal entity identifiers](/v1/docs/representing-institutions-in-lusid-using-legal-entities#understanding-userdefined-identifiers) |
| **Display name source** and **Description source** | The record fields to use as the legal entity’s display name and description respectively |
| **Additional identifiers** | Additional identifiers to write alongside the primary identifier, as described [here](/v1/docs/representing-institutions-in-lusid-using-legal-entities#understanding-userdefined-identifiers) |
| **Property mappings** | Each property mapping configures the processor to decorate a LUSID property onto legal entities with a value from your specified source field |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as legal entities in LUSID |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertLegalEntities",
  "Config": {
    "identifierScope": "Finbourne-Examples",
    "identifierType": "MancoCode",
    "identifierValueSource": "MancoCode",
    "displayNameSource": "EntityName",
    "descriptionSource": "EntityDescription",
    "additionalIdentifiers": [{ "Scope": "Finbourne-Examples", "Type": "Lei", "SourceField": "LEI" }],
    "propertyMappings": [
      { "PropertyKey": "LegalEntity/Finbourne-Examples/Country", "SourceField": "Country", "DataType": "string", "Unit": null }
    ],
    "batchSize": 100,
    "continueAfterSink": false
  }
}
```

## Upsert property

The **Upsert property** processor [adds properties](/v1/docs/how-do-i-add-a-property-to-an-entity) onto existing LUSID entities. The processor updates or adds to the entity’s existing properties rather than replacing them.

| Setting | Description |
| --- | --- |
| **Entity type** | The LUSID entity type to write properties onto |
| **<Entity type> source** | The record field to source the required identifying data from, for example portfolio scope and code, or transaction ID |
| **Property mappings** | **Data type** | How the processor should interpret and standardise each value before upserting it as a property value: - `date` and `datetime`: Formats values as ISO-8601 UTC - `collection`: Identifies and splits values in a [multi-value property](/v1/docs/understanding-multi-value-properties) on your specified **Delimiter** |
| **Unit source field** | A record field to source the required metric unit from, for example `Ccy`, when processing the following data types: - `number` - `decimal` - `metric` |
| **Unit** | Hardcode a specific metric unit, for example `USD`, to include when processing the following data types: - `number` - `decimal` - `metric` |
| **Value** | Hardcode a specific value for the property; this takes precedence if you specify both a **Value** and a **Source field** |
| **Source field** | The record field to source the property value from |
| **Effective date source field** | The record field to source the property value’s [effectiveFrom](/v1/docs/what-are-the-effectiveat-and-effectivefrom-timestamps) date from. Leave blank for perpetual properties (or to set time-variant properties to `0000-00-00`) |
| **Delimiter** | Identifies the separator for multi-value properties when processing the `collection` data type |
| **Property key** | An existing [three-stage property key](/v1/docs/properties) to decorate onto entities along with the configured values |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as properties in LUSID |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertProperty",
  "Config": {
    "entityType": "Instrument",
    "identifierType": "Isin",
    "identifierSource": "ISIN",
    "scope": "Finbourne-Examples",
    "propertyMappings": [
      { "propertyKey": "Instrument/Finbourne-Examples/Rating", "sourceField": "Rating", "dataType": "string" },
      { "propertyKey": "Instrument/Finbourne-Examples/Yield", "sourceField": "Yield", "dataType": "number", "unit": "Percent" }
    ],
    "batchSize": 100,
    "continueAfterSink": false
  }
}
```

## Upsert property definitions

The **Upsert property definitions** processor creates or updates LUSID property definitions. Once a property definition has been created in LUSID, you can use the [**Upsert property** processor](/v1/docs/processor-types-for-custom-dataflow-integrations#upsert-property) to decorate that property onto entities.

> [!NOTE]
> Note
> 
> The processor updates an existing LUSID property definition if it matches one of the supplied three-stage property keys (`&lt;Domain&gt;/&lt;Scope&gt;/&lt;Code&gt;`). The processor can only update the following property definition fields:
> 
> - `displayName`
> - `propertyDescription`

| Setting | Description |
| --- | --- |
| **Property definitions** | Use hardcoded values to [create property definitions](/v1/docs/how-do-i-create-a-property-type) |
| **Dynamic property definitions** | Source values for the property definition `Code` from the specified record field |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as property definitions in LUSID |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertPropertyDefinition",
  "Config": {
    "propertyDefinitions": [
      {
        "domain": "Instrument",
        "scope": "MyVendor",
        "code": "Rating",
        "displayName": "Credit Rating",
        "dataTypeScope": "system",
        "dataTypeCode": "string",
        "description": "Vendor-supplied credit rating",
        "lifetime": "TimeVariant",
        "constraintStyle": "Property"
      }
    ],
    "continueAfterSink": false
  }
}
```

## Upsert quote

The **Upsert quote** processor creates or updates one or more [quotes](/v1/docs/how-do-i-upload-a-market-price-or-fx-spot-rate-to-the-quote-store) in LUSID for each record. A single record can produce multiple quotes, for example one quote each for `bid`, `mid`, and `ask`.

LUSID uses each quote’s unique combination of the following fields to determine whether an incoming quote is new or an update to an existing quote:

- Provider
- Price source
- Instrument identifier
- Identifier type
- Quote type
- Field
- Effective date

| Setting | Description |
| --- | --- |
| **<Quote field>** | Use hardcoded values to create or update quotes; see [How do I load market prices into the Quote Store?](/v1/docs/how-do-i-upload-a-market-price-or-fx-spot-rate-to-the-quote-store#data-fields-and-properties) for information on each field |
| **Instrument ID source** and **Instrument ID type** | The record field to source the quote’s instrument ID from, and its [identifier type](/v1/docs/understanding-instrument-identifiers) |
| **EffectiveAt source** | The record field to source the quote’s [effectiveAt](/v1/docs/what-are-the-effectiveat-and-effectivefrom-timestamps) date from |
| **EffectiveAt format** | See [Reference: Date formats](/v1/docs/processor-types-for-custom-dataflow-integrations#reference-date-formats) |
| **Timezone** | The [time zone](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones#List) of date values |
| **Quote field** and **Source field** | LUSID upserts one quote per price mapping pair (if a source value is found): - **Quote field**: A [supported field for the chosen provider](/v1/docs/understanding-valid-market-data-provider-and-field-combinations), for example `bid`, `mid`, or `ask` - **Source field**: The record field to source the price value from |
| **Currency source** or **Currency** | Hardcode the currency of each quote, or specify the record field to source each quote’s currency from. If both are set, **Currency source** takes precedence. |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as quotes in LUSID |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertQuote",
  "InputFlows": { "default": "priced" },
  "Config": {
    "scope": "MyQuoteScope",
    "provider": "MyProvider",
    "priceSource": "MLPD",
    "instrumentIdSource": "_ResolvedLuid",
    "instrumentIdType": "LusidInstrumentId",
    "effectiveAtSource": "PriceDate",
    "effectiveAtFormat": "yyyy-MM-dd",
    "dateTimeZone": "America/Toronto",
    "priceMappings": [
      { "field": "Bid", "source": "BidPrice" },
      { "field": "Mid", "source": "MidPrice" },
      { "field": "Ask", "source": "AskPrice" }
    ],
    "quoteType": "Price",
    "currencySource": "Ccy",
    "currencyLiteral": "USD",
    "lineage": "MyProvider_BatchRun_{batchRunId}_{{DATE:T:yyyy-MM-ddTHHmmssZ}}",
    "scaleFactor": 1,
    "batchSize": 2000,
    "continueAfterSink": false
  }
}
```

## Upsert reference portfolio

The **Upsert reference portfolio** processor creates or updates [reference portfolios](/v1/docs/what-is-a-reference-portfolio) from records. If a reference portfolio with the same scope and code already exists, the processor updates it; otherwise the processor creates it.

| Setting | Description |
| --- | --- |
| **Scope** | The LUSID [scope](/v1/docs/what-is-a-scope) to create or update reference portfolios in |
| **<Reference portfolio field> source** | The record field to populate each reference portfolio field value from; [read more on LUSID’s reference portfolio fields](/v1/docs/how-do-i-create-a-reference-portfolio) |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as reference portfolios in LUSID |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertReferencePortfolio",
  "Config": {
    "scope": "Finbourne-Examples",
    "portfolioCodeSource": "IndexCode",
    "displayNameSource": "IndexName",
    "descriptionSource": "IndexDescription",
    "baseCurrencySource": "Currency",
    "batchSize": 200,
    "recordTypeFilter": "IndexHeader",
    "continueAfterSink": false
  }
}
```

## Upsert reference portfolio constituent

The **Upsert reference portfolio constituent** processor creates or updates the [constituents of reference portfolios](/v1/docs/loading-constituents-and-rebalancing-a-reference-portfolio) from records. The processor groups records by portfolio code and [effectiveAt](/v1/docs/what-are-the-effectiveat-and-effectivefrom-timestamps) date, and then upserts each group. This means each constituent upsert replaces the complete constituent set for that portfolio and date.

| Setting | Description |
| --- | --- |
| **Scope** | The LUSID [scope](/v1/docs/what-is-a-scope) to create or update reference portfolio constituents in |
| **<Constituent field> source** | The record field to populate each reference portfolio field value from; [read more on LUSID’s reference portfolio constituent fields](/v1/docs/loading-constituents-and-rebalancing-a-reference-portfolio) |
| **EffectiveAt format** | See [Reference: Date formats](/v1/docs/processor-types-for-custom-dataflow-integrations#reference-date-formats) |
| **LUSID identifier type** and **Source field** | The record field to source the constituent’s instrument ID from, and its hardcoded [identifier type](/v1/docs/understanding-instrument-identifiers) |
| **Property mappings** | Each property mapping configures the processor to decorate a LUSID property onto constituents with a value from your specified source field |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as reference portfolio constituents in LUSID |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertReferencePortfolioConstituent",
  "Config": {
    "scope": "Finbourne-Examples",
    "portfolioCodeSource": "IndexCode",
    "effectiveAtSource": "EffectiveDate",
    "effectiveAtFormat": "yyyy-MM-dd",
    "instrumentIdentifierMappings": [
      { "lusidIdentifierType": "Instrument/default/Isin",  "sourceField": "Isin",  "isRequired": true },
      { "lusidIdentifierType": "Instrument/default/Cusip", "sourceField": "Cusip", "isRequired": false }
    ],
    "weightSource": "Weight",
    "currencySource": "Currency",
    "propertyMappings": [
      { "propertyKey": "ReferenceHolding/IndexProvider/Sector", "sourceField": "Sector", "dataType": "string" }
    ]
  }
}
```

## Upsert relational dataset

The **Upsert relational dataset** processor creates or updates data points in a [LUSID relational dataset](/v1/docs/custom-datasets) from records. If the dataset definition does not yet exist in your domain, the processor creates it automatically on the first run.

| Setting | Description |
| --- | --- |
| **Scope** and **Code** | The [relational dataset definition](/v1/docs/how-do-i-create-a-relational-dataset-definition) scope and code |
| **Display name** and **Description** | A friendly display name and description for the dataset |
| **Create if not exists** | When enabled, creates the relational dataset definition automatically on the first run |
| **Column definitions** | The [field schema](/v1/docs/how-do-i-create-a-relational-dataset-definition#request-fields) for the relational dataset definition: **Key columns** become series identifiers; non-key columns become value fields. See [Anatomy of a dataset](/v1/docs/anatomy-of-a-dataset). |
| **Entity type** | The type of entity to associate each dataset record with; see [applicableEntityTypes](/v1/docs/how-do-i-create-a-relational-dataset-definition#request-fields) |
| **Column mappings** | Where to source the values for each data point field; see [How do I add records to a dataset?](/v1/docs/how-do-i-add-records-to-a-custom-dataset#request-fields) |
| **Data series scope** | A series scope for each record; see [Constructing the primary key](/v1/docs/how-do-i-add-records-to-a-custom-dataset#constructing-the-primary-key) |
| **EffectiveAt source** | The record field to source the effectiveAt datetime from |
| **EffectiveAt default** | If both **EffectiveAt default** and **source** are not set, defaults to the current UTC time |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as dataset records in LUSID |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertRelationalDataSet",
  "Config": {
    "scope": "Fitch",
    "code": "IssuerRatings",
    "createIfNotExists": true,
    "displayName": "Fitch Issuer Ratings",
    "applicableEntityTypes": ["LegalEntity"],
    "entityType": "LegalEntity",
    "dataSeriesScope": "Fitch",
    "effectiveAtSource": "RatingDate",
    "effectiveAtDefault": "2026-01-01",
    "columnDefinitions": [
      { "Name": "RatingType", "Type": "string", "IsKey": true, "IsRequired": true, "Category": "SeriesIdentifier" },
      { "Name": "Rating", "Type": "string", "IsKey": false, "IsRequired": true, "Category": "Value" },
      { "Name": "Source", "Type": "string", "IsKey": false, "IsRequired": false, "Category": "Metadata" }
    ],
    "columnMappings": [{ "DataSetColumn": "Rating", "SourceField": "LongTermRating" }],
    "entityIdentifierScope": "default",
    "entityIdentifierType": "FitchEntityId",
    "entityIdentifierSource": "IssuerId",
    "batchSize": 500
  }
}
```

## Upsert settlement instructions

*More information coming soon.*

## Upsert transaction

The **Upsert transaction** processor creates or updates transactions in LUSID from records. The processor groups and upserts records by portfolio.

| Setting | Description |
| --- | --- |
| **Portfolio scope** | The LUSID portfolio scope to write transactions into |
| **Portfolio code source** | The record field to source the portfolio code from |
| **<Transaction field> source** | The record field to populate each value on a transaction from; [read more on LUSID’s transaction fields](/v1/docs/how-do-i-create-or-update-a-transaction) |
| **Property mappings** | Each property mapping configures the processor to decorate a LUSID property onto transactions with a value from your specified source field |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after upserting them as transactions in LUSID |
| **Record type filter** | If specified, only process records with this record type attached |
| **Pass through unmatched** | Passes records that do not match the **Record type filter** downstream unchanged (rather than dropping them) |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "UpsertTransaction",
  "InputFlows": { "resolved": "resolved", "created": "created" },
  "Config": {
    "scope": "Finbourne-Examples",
    "portfolioCodeSource": "_ResolvedPortfolioCode",
    "transactionIdSource": "TradeId",
    "typeSource": "TxnType",
    "typeMap": { "BUY": "Buy", "SELL": "Sell" },
    "transactionDateSource": "TradeDate",
    "transactionDateFormat": "yyyyMMdd",
    "settlementDateSource": "SettleDate",
    "dateTimeZone": "America/New_York",
    "unitsSource": "Quantity",
    "priceSource": "Price",
    "totalConsiderationAmountSource": "NetAmount",
    "totalConsiderationCurrencySource": "Ccy",
    "exchangeRateSource": "FxRate",
    "source": "Custodian",
    "instrumentIdentifierSource": "_ResolvedLuid",
    "identifierMappings": [
      { "lusidIdentifierType": "Isin", "sourceField": "ISIN" },
      { "lusidIdentifierType": "Cusip", "sourceField": "CUSIP" }
    ],
    "propertyMappings": [
      { "propertyKey": "Transaction/Finbourne-Examples/Strategy", "sourceField": "Strategy", "dataType": "string" }
    ],
    "custodianAccountScope": "Custodians",
    "custodianAccountCodeSource": "_ResolvedCustodianAccountCode",
    "batchSize": 100,
    "continueAfterSink": false
  }
}
```

## Publish Kafka

The **Publish Kafka** processor publishes each record to a Kafka topic as a JSON message.

> [!NOTE]
> Note
> 
> To read from a Kafka topic, use the [Ingest](/v1/docs/processor-types-for-custom-dataflow-integrations#ingest) processor with the **Source type** set to `ActiveMQ`.

| Setting | Description |
| --- | --- |
| **Topic** | The Kafka topic to publish to |
| **Key source field** | The record field to source the Kafka message key from |
| **Value format** | Currently supports JSON |
| **Bootstrap servers** | The broker list, for example `host1:1234`, `host2:5678` |
| **Credentials key** | A prefix used to look up bootstrap servers and SASL credentials from the [Configuration Store](/v1/docs/configuration-store), for example `{key}-bootstrapServers`, `{key}-saslUsername`, `{key}-saslPassword` |
| **Additional config** | Key/value pairs to pass in extra Kafka client settings, for example `compression.type` or `acks` |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after publishing them as Kafka messages |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

**Configuration store credentials**

```json
{
  "ProcessorType": "PublishKafka",
  "InputFlows": { "default": "enriched" },
  "Config": {
    "topic": "positions.outbound",
    "credentialsKey": "client-kafka",
    "keySourceField": "PortfolioCode",
    "batchSize": 1000,
    "additionalConfig": {
      "compression.type": "gzip"
    }
  }
}
```

**Inline broker list**

```json
{
  "ProcessorType": "PublishKafka",
  "InputFlows": { "default": "trades" },
  "OutputFlows": { "default": "published" },
  "Config": {
    "topic": "trades.outbound",
    "bootstrapServers": "broker1:1234,broker2:5678",
    "keySourceField": "TradeId",
    "continueAfterSink": true
  }
}
```

## Publish RabbitMQ

The **Publish RabbitMQ** processor publishes each record to a RabbitMQ exchange as a JSON message. The processor waits for a per-record broker acknowledgement before recording a success outcome for each message.

| Setting | Description |
| --- | --- |
| **Credentials key** (recommended) or **Connection URI** | Hardcode the AMQP or AMQPS connection URI (supports `SECRET:` tokens), or specify a **Credentials key** prefix to look up the connection URI, username, and password from the [Configuration Store](/v1/docs/configuration-store), for example `{key}-username`, `{key}-password`, `{key}-connectionUri` |
| **Host name override**, **Port override**, **Virtual host override** | Optional overrides to apply on top of the resolved connection URI |
| **Exchange** | The target exchange name |
| **Routing key** or **Routing key source field** | Hardcode the routing key for every message, or specify the record field to source the per-message routing key from |
| **Declare exchange** and **Exchange type** | For testing purposes; when enabled, declares the exchange before publishing |
| **Persistent delivery** | When enabled, publishes messages as persistent (delivery mode 2); requires the exchange and queue to be durable |
| **Value format** | Currently supports JSON |
| **Continue after sink** | When enabled, the processor passes the records to the next processor after publishing them as RabbitMQ messages |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "PublishRabbitMq",
  "Config": {
    "credentialsKey": "rmq-credentials",
    "exchange": "finbourne-examples.instruments",
    "routingKeySourceField": "AssetClass",
    "persistentDelivery": true,
    "batchSize": 500
  }
}
```

## Notification

The **Notification** processor sends an email [notification](/v1/docs/how-do-i-notify-people-via-email-when-an-event-occurs) at the end of a dataflow run. You can choose whether to send on success, failure, or always, and attach a summary of what the run processed.

| Setting | Description |
| --- | --- |
| **SMTP server** | The hostname, port, username, and password for your SMTP server |
| **Use SSL/TLS** | Disable this for local testing tools only |
| **From** | The sender email address |
| **To** | Up to 10 direct recipient email addresses |
| **CC** | Up to 10 carbon copy recipient email addresses |
| **BCC** | Up to 10 blind carbon copy recipient email addresses |
| **Subject** | The email subject line; you can use template tokens to make the text more meaningful |
| **Body** | The email body as plain text or HTML; supports template tokens |
| **HTML email** | When enabled, the processor sends the email as formatted HTML instead of plain text |
| **Send condition** | When to send this notification based on the run state of the dataflow instance: `Always`, `OnSuccess`, or `OnFailure` |
| **Attach flow summary** | When enabled, attaches a summary of the run’s record counts to the email as a CSV |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "Notification",
  "Config": {
    "NotificationType": "SendEmail",
    "smtpHost": "smtp.acme.com",
    "smtpPort": 587,
    "useSsl": true,
    "username": "{{SECRET:smtp-user}}",
    "password": "{{SECRET:smtp-pass}}",
    "from": "user@acme.com",
    "to": ["ops@acme.com"],
    "cc": [],
    "subject": "Dataflow {{Status}} — run {{RunId}}",
    "body": "Processed {{RecordCount}} records with {{ErrorCount}} errors on {{ExecutionDate}}.",
    "isHtml": false,
    "sendCondition": "Always",
    "attachFlowSummary": true
  }
}
```

## Workflow

The **Workflow** processor creates tasks in the [Workflow Service](/v1/docs/workflow-service) from records - one task per record. You can filter which records trigger a task, assign tasks to specific users or teams, and map record fields onto task fields.

| Setting | Description |
| --- | --- |
| **Task definition scope** and **Task definition code** | The [scope and code of the task definition](/v1/docs/how-do-i-create-a-task-definition) to kick off a task from |
| **Initial state** | Set the workflow state for new tasks |
| **Due date offset** | Sets the task due date to the run date plus this number of days and hours |
| **Condition** | A LUSID filtering syntax expression that evaluates to true or false; records that do not evaluate to true pass through the processor with no task created |
| **Record type filter** | If specified, only process records with this record type attached |
| **Field mappings** | Where to source the values for each [task field](/v1/docs/how-do-i-create-a-task) |

For example, you could include the following segment in [your imported JSON template](/v1/docs/creating-a-custom-dataflow-integration-from-a-template) to automatically populate the processor settings:

```json
{
  "ProcessorType": "Workflow",
  "Config": {
    "when": "Status == 'BREAK'",
    "recordTypeFilter": "Reconciliation",
    "taskSequence": [
      {
        "taskDefinitionScope": "Finbourne-Examples",
        "taskDefinitionCode": "review-break",
        "initialState": "Open",
        "dueDateOffsetDays": 2,
        "assignment": { "mode": "static", "assignee": "ops-team" },
        "fieldMappings": [
          { "taskField": "PortfolioCode", "sourceField": "Portfolio", "defaultValue": null },
          { "taskField": "Amount", "sourceField": "BreakAmount", "defaultValue": "0" }
        ]
      },
      {
        "taskDefinitionScope": "ops",
        "taskDefinitionCode": "approve-resolution",
        "initialState": "Pending",
        "dependsOnPrevious": true,
        "assignment": { "mode": "roundRobin", "pool": ["joeBloggs", "janeSmith"] }
      }
    ]
  }
}
```

For bitemporal data, the **effectiveAt** timestamp reflects the point in time at which a data item becomes valid in LUSID from a business perspective. You can specify the field when you make a read or update request to LUSID, indicating that you want to retrieve or update data that is effective at the specified date.

An occurrence of the workflow defined in a task definition. You can create multiple tasks from one task definition, typically with changing input values, for example you might create one task per LUID for securities to be screened.

A condition a task can exist in, for example **Pending**, **In progress**, or **Complete**. State names are customisable and can be defined in a task definition.