---
title: "Processor types for custom dataflow integrations"
slug: "processor-types-for-custom-dataflow-integrations"
updated: 2026-07-20T15:11:30Z
published: 2026-07-20T15:11:30Z
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 dataflows 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)

### 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)

#### 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` |

## 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` |

## 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

*More information is coming soon.*

## Rename

*More information is coming soon.*

## 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 pipe - **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) |

## 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` |

## 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 |

## 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 |

## 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) |

## 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 |

## Upsert property

*More information is coming soon.*

## Upsert property definitions

*More information is coming soon.*

## Upsert quote

*More information is coming soon.*

## Upsert reference portfolio

*More information is coming soon.*

## Upsert reference portfolio constituents

*More information is coming soon.*

## 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 |

## 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) |

## Publish Kafka

*More information is coming soon.*

## Publish RabbitMQ

*More information is coming soon.*

## Notification

The **Notification** processor sends an email notification 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 |

## 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 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.