Restructure JSON or hash objects

Overview

By default, RDI adds fields to
[hash]() or
[JSON]() objects in the target
database that closely match the columns of the source table.
If you just want to limit the set fields in the output and/or rename some of them, you can use the
[`output mapping`]() configuration option.

For situations where you want to create a new object structure with multiple levels or use calculations for the field values, you can use the
[`map`]()
transformation, as described in the following sections.

Creating multilevel JSON objects

You can use the `map` transformation to create a new structure for the output data, which can include nested objects and calculated fields. The `map` transformation allows you to define a new structure using an expression language, such as SQL or JavaScript.

[code example]

The example above creates a new JSON object with the following structure:
 - A top-level `id` field that is the same as the `employeeid` field in the source table.
 - A `name` field that is a concatenation of the `firstname` and `lastname` fields, with the `lastname` converted to uppercase.
 - An `address` subobject that contains the `address`, `city`, `state`, `postalcode`, and `country` fields.
 - A `contact` subobject that contains the `phone` field and a modified version of the `email` field, where the '@' sign and dots are replaced with '_at_' and '_dot_' respectively.

The `output` section of the file configures the job to write
to a JSON object with a custom key. Note that in the `output` section, you must refer to
fields defined in the `map` transformation, so we use the new name `id`
for the key instead of `employeeid`.

If you query one of the new JSON objects, you see output like the following:

[code example]

Formatted in the usual JSON style, the output looks like the sample below:

[code example]

Creating hash structure

This example creates a new [hash]()
object structure for items from the `track` table. Here, the `map` transformation uses
[SQL](https://en.wikipedia.org/wiki/SQL) for the expression because this is often
more suitable for hashes or "flat"
JSON objects without subobjects or arrays. The expression renames some of the fields.
It also calculates more human-friendly representations for the track duration (originally
stored in the `milliseconds` field) and the storage size (originally stored in the
`bytes` field).

The full example is shown below:

[code example]

If you query the data for one of the `track` hash objects, you see output
like the following:

[code example]

name: Create multilevel employee JSON
source:
  db: chinook
  table: employee

transform:
  - uses: map
    with:
      expression: |
        {
          "id": employeeid,
          "name": concat([firstname, ' ', upper(lastname)]),
          "address": {
            "street": address,
            "city": city,
            "state": state,
            "postalCode": postalcode,
            "country": country
          },
          "contact": {
            "phone": phone,
            "safeEmail": replace(replace(email, '@', '_at_'), '.', '_dot_')
          }
        }
      language: jmespath

output:
  - uses: redis.write
    with:
      data_type: json
      key:
        expression: concat(['emp:', id])
        language: jmespath

> JSON.GET emp:1 $
"[{\"id\":1,\"name\":\"Andrew ADAMS\",\"address\":{\"street\":\"11120 Jasper Ave NW\",\"city\":\"Edmonton\",\"state\":\"AB\",\"postalCode\":\"T5K 2N1\",\"country\":\"Canada\"},\"contact\":{\"phone\":\"+1 (780) 428-9482\",\"safeEmail\":\"andrew_at_chinookcorp_dot_com\"}}]"

name: Create track hash with calculated fields
source:
  db: chinook
  table: track
transform:
  - uses: map
    with:
      expression:
          id: trackid
          name: name
          duration: concat(floor(milliseconds / 60000), ':', floor(mod(milliseconds / 1000, 60)))
          storagesize: concat(round(bytes / 1048576.0, 2), 'MB')
      language: sql
output:
  - uses: redis.write
    with:
      connection: target
      data_type: hash
      key:
        expression: concat('track:', id)
        language: sql

> hgetall track:16
1) "id"
2) "16"
3) "name"
4) "Dog Eat Dog"
5) "duration"
6) "3:35.0"
7) "storagesize"
8) "6.71MB"


Add new fields to a key

Overview

By default, RDI adds fields to
[hash]() or
[JSON]() objects in the target
database that match the columns of the source table.
The examples below show how to add extra fields to the target data with the
[`add_field`]() transformation.

Add a single field

The first example adds a single field to the data.
The `source` section selects the `customer` table of the
[`chinook`](https://github.com/Redislabs-Solution-Architects/rdi-quickstart-postgres)
database (the optional `db` value here corresponds to the
`sources.<source-name>.connection.database` value defined in
[`config.yaml`]()).

In the `transform` section, the `add_field` transformation adds an extra field called `localphone`
to the object, which is created by removing the country and area code from the `phone`
field with the
[JMESPath]() function `regex_replace()`.
You can also specify `sql` as the `language` if you prefer to create the new
field with an [SQL](https://en.wikipedia.org/wiki/SQL) expression.

The `output` section specifies `hash` as the `data_type` to write to the target, which
overrides the default setting of `target_data_type` defined in `config.yaml`. Also, the
`output.with.key` section specifies a custom key format of the form `cust:<id>` where
the `id` part is generated by the `uuid()` function.

The full example is shown below:

[code example]

If you queried the generated target data from the default transformation
using [`redis-cli`](), you would
see something like the following:

[code example]

Using the job file above, the data also includes the new `localphone` field:

[code example]

Add multiple fields

The `add_field` transformation can also add multiple fields at the same time
if you specify them under a `fields` subsection. The example below adds two
fields to the `track` objects. The first new field, `seconds`, is created using a SQL
expression to calculate the duration of the track in seconds from the
`milliseconds` field.
The second new field, `composerlist`, adds a JSON array using the `split()` function
to split the `composer` string field wherever it contains a comma.

[code example]

You can query the target database to see the new fields in
the JSON object:

[code example]

Using `add_field` with `remove_field`

You can use the `add_field` and
[`remove_field`]()
transformations together to completely replace fields from the source. For example,
if you add a new `fullname` field, you might not need the separate `firstname` and
`lastname` fields. You can remove them with a job file like the following:

[code example]

name: Add local phone field to customer
source:
  db: chinook
  table: customer
transform:
  - uses: add_field
    with:
      expression: regex_replace(phone, '\+[0-9]+ (\([0-9]+\) )?', '')
      field: localphone
      language: jmespath
output:
  - uses: redis.write
    with:
      connection: target
      data_type: hash
      key:
        expression: concat(['cust:', uuid()])
        language: jmespath

1) "customerid"
 2) "27"
 3) "firstname"
 4) "Patrick"
 5) "lastname"
 6) "Gray"
 .
 .
23) "localphone"
24) "622-4200"

name: Add multiple fields to track
source:
  db: chinook
  table: track
transform:
  - uses: add_field
    with:
      fields:
        - expression: floor(milliseconds / 1000)
          field: seconds
          language: sql
        - expression: split(composer)
          field: composerlist
          language: jmespath
output:
  - uses: redis.write
    with:
      connection: target
      data_type: json
      key:
        expression: concat(['track:', trackid])
        language: jmespath

> JSON.GET track:1 $

"[{\"trackid\":1,\"name\":\"For Those About To Rock (We Salute You)\",\"albumid\":1,\"mediatypeid\":1,\"genreid\":1,\"composer\":\"Angus Young, Malcolm Young, Brian Johnson\",\"milliseconds\":343719,\"bytes\":11170334,\"unitprice\":\"0.99\",\"seconds\":343,\"composerlist\":[\"Angus Young\",\" Malcolm Young\",\" Brian Johnson\"]}]"

name: Add fullname and remove separate name fields
source:
  db: chinook
  table: customer
transform:
  - uses: add_field
    with:
      expression: concat(firstname, ' ', lastname)
      field: fullname
      language: sql
  - uses: remove_field
    with:
      fields:
        - field: firstname
        - field: lastname
output:
  - uses: redis.write
    with:
      connection: target
      data_type: hash
      key:
        expression: concat(['cust:', customerid])
        language: jmespath


Set custom expiration times / TTL

Overview

You can configure custom key expiration times (TTL) for keys written to Redis by using the `expire` parameter in the `output` section of the job file. This parameter specifies the duration, in seconds, that a newly created key will remain in Redis before being automatically deleted. If the `expire` parameter is not provided, the keys will persist indefinitely.

There are two ways to set the expiration time:

- as a static value
- as a dynamic value using a JMESPath or SQL expression

Static expiration time

The following example sets the expiration time to 100 seconds for all keys:

[code example]

Dynamic expiration time

You can use a JMESPath or SQL expression to set the expiration time dynamically when it is based on a field in the source data. For example, you can set the expiration time to the value of a `ttl` field in the source data:

[code example]

Dynamic expiration time based on a date, datetime, or timestamp field

In some cases, you can also set the expiration time based on a field that contains a date, datetime, or timestamp value, but it depends on the source database and the data types it supports. See the examples below for your specific source database and data type.

There are two main approaches you can use to set the expiration time based on a date, datetime, or timestamp field:

- For values representing an elapsed time since epoch start (in milliseconds, for example), you have to convert the value to seconds since epoch and then subtract the current time (also in seconds since epoch). The difference between the two is the time until expiration.

    [code example]

- For values matching the subset of ISO 8601 supported by SQLite (for example, `2023-10-01T12:00:00`, `2023-10-01T12:00:00Z`, or `2025-06-05T13:40:14.784000+02:00`), you can use the `STRFTIME` function to convert the value to seconds since epoch and subtract the current time in seconds since epoch from it.

  [code example]

For more examples of how to manipulate date and time values, see [Formatting date and time values]().

name: Static expiration example
output:
  - uses: redis.write
    with:
      data_type: hash
      expire: 100

name: Dynamic expiration from field
output:
  - uses: redis.write
    with:
      data_type: hash
      expire:
        expression: ttl
        language: jmespath

name: Expiration from timestamp in milliseconds
    output:
        - uses: redis.write
          with:
            data_type: hash
            expire:
              # To set the expiration time to a date field, convert the value to
              # seconds (e.g. divide it by 1000 if the fields has milliseconds precision)
              # and subtract the current time in seconds since epoch.
              expression: EXPIRES_TIMESTAMP / 1000 - STRFTIME('%s', 'now')
              language: sql

name: Expiration from ISO 8601 datetime
  output:
    - uses: redis.write
      with:
        data_type: hash
        expire:
          language: sql
          expression: STRFTIME('%s', EXPIRATION_TS) - STRFTIME('%s', 'now')


Write to a Redis hash

Content

In the following example, the data is captured from the source table named `invoice` and is written to the Redis database as hash keys. The `connection` is an optional parameter that refers to the corresponding connection name defined in `config.yaml`.
When you specify the `data_type` parameter for the job, it overrides the system-wide setting `target_data_type` defined in `config.yaml`.

In this case, the result will be Redis hashes with key names based on the key expression (for example, `invoice_id:1`) and with an expiration of 100 seconds.
If you don't supply an `expire` parameter, the keys will never expire.

[code example]

name: Write invoice to hash
source:
  schema: public
  table: invoice
output:
  - uses: redis.write
    with:
      connection: target
      data_type: hash
      key:
        expression: concat(['invoice_id:', InvoiceId])
        language: jmespath
      expire: 100


Write to a Redis JSON document

Content

You must enable the [RedisJSON]() module in the target Redis
database to use this feature.

In the example below, the data is captured from the source table named `invoice` and is written to the Redis database as a JSON document. The `connection` is an optional parameter that refers to the corresponding connection name defined in `config.yaml`. When you specify the `data_type` parameter for the job, it overrides the system-wide setting `target_data_type` defined in `config.yaml`.

Another optional parameter, `on_update`, specifies the writing strategy. You can set this to either `replace` (the default) or `merge`. This affects the way the document is written to the target. Replacing the document will overwrite it completely, while merging will update it with the fields captured in the source, keeping the rest of the document intact. The `replace` option is usually more performant, while `merge` allows other jobs and applications to set extra fields in the same JSON documents.

In this case, the result will be Redis JSON documents with key names based on the key expression (for example, `invoice_id:1`) and with an expiration of 100 seconds. If you don't supply an `expire` parameter, the keys will never expire.

[code example]

name: Write invoice to JSON
source:
  schema: public
  table: invoice
output:
  - uses: redis.write
    with:
      connection: target
      data_type: json
      key:
        expression: concat(['invoice_id:', InvoiceId])
        language: jmespath
      on_update: replace
      expire: 100


Using the operation code

Overview

The operation code (`opcode`) is a metadata field that indicates the type of operation that generated the change in the source database. It can be useful for tracking changes and understanding the context of the data being processed.

The opcode is only available in the [full row format](), and can be accessed in the `transform` and `output` sections of the job file.

It has one of the following values:

- r - Read (applies to only snapshots)
- c - Create
- u - Update
- d - Delete
- t = Truncate (PostgreSQL specific)
- m = Message (PostgreSQL specific)

You can add the value of the operation code to the output, and also use it in a conditional expression to modify the behavior of the job. The following examples demonstrate the different use-cases.

Adding the operation code to the output

Use the `add_field` transformation to add a new field that contains the value of the `opcode` field from the source data. Note that the fields must be prefixed with `after` to be included in the output.

[code example]

Filtering operation by output code.

In some cases you may want to ignore certain operations (for example, you may not be interested in deletions). Use the `filter` transformation to filter out any operations you don't need to process.

[code example]

Modifying the output based on the operation code

The previous example filters out specific operations, but you can also modify the output based on the operation code. For example, you can add a new field that tracks the status of the record based on the operation code.

Note that when a source record is deleted, you must modify the value of the `opcode` field if you want to prevent the corresponding record in the target database from being removed automatically.

[code example]

name: Add operation code to employee
source:
  schema: public
  table: employee
  row_format: full

transform:
  # add the operation code to the data
  - uses: add_field
    with:
      field: after.operation_code
      expression: opcode
      language: jmespath

name: Filter out delete operations
source:
  schema: public
  table: employee
  row_format: full

transform:
  - uses: filter
    with:
      expression: opcode != 'd'
      language: jmespath

name: Track status based on operation code
source:
  schema: public
  table: employee
  row_format: full

transform:
  - uses: add_field
    with:
      fields:
        # Here you set the value of the field based on the value of the opcode field
        - field: after.status
          expression: opcode == 'd' && 'inactive' || 'active'
          language: jmespath

        # You have to change the value of the opcode field to prevent deletion
        - field: opcode
          expression: opcode == 'd' && 'u' || opcode
          language: jmespath


Remove fields from a key

Overview

By default, RDI adds fields to
[hash]() or
[JSON]() objects in the target
database for each of the columns of the source table.
The examples below show how to omit some of those fields from the target data with the
[`remove_field`]() transformation.

Remove a single field

The first example removes a single field from the data.
The `source` section selects the `employee` table of the
[`chinook`](https://github.com/Redislabs-Solution-Architects/rdi-quickstart-postgres)
database (the optional `db` field here corresponds to the
`sources.<source-name>.connection.database` field defined in
[`config.yaml`]()).

In the `transform` section, the `remove_field` transformation removes the
`hiredate` field.

The `output` section specifies `hash` as the `data_type` to write to the target, which
overrides the default setting of `target_data_type` defined in `config.yaml`. Also, the
`output.with.key` section specifies a custom key format of the form `emp:<employeeid>`.
Note that any fields you remove in the `transform` section are not available for
the key calculation in the `output` section.

The full example is shown below:

[code example]

If you queried the generated target data from the default transformation
using [`redis-cli`](), you would
see something like the following:

[code example]

Using the job file above, the data omits the `hiredate` field:

[code example]

Remove multiple fields

The `remove_field` transformation can also remove multiple fields at the same time
if you specify them under a `fields` subsection. The example below is similar
to the previous one but also removes the `birthdate` field:

[code example]

If you query the data, you can see that it also omits the
`birthdate` field:

[code example]

Using `remove_field` with `add_field`

The `remove_field` transformation is very useful in combination with
[`add_field`]().
For example, if you use `add_field` to concatenate a person's first
and last names, you may not need separate `firstname` and `lastname`
fields, so you can use `remove_field` to omit them.
See [Using `add_field` with `remove_field`]()
for an example of how to do this.

name: Remove hiredate from employee
source:
  db: chinook
  table: employee
transform:
  - uses: remove_field
    with:
      field: hiredate
output:
  - uses: redis.write
    with:
      connection: target
      data_type: hash
      key:
        expression: concat(['emp:', employeeid])
        language: jmespath

> hgetall emp:8
 1) "employeeid"
 2) "8"
 3) "lastname"
 4) "Callahan"
 5) "firstname"
 6) "Laura"
 7) "title"
 8) "IT Staff"
 9) "reportsto"
10) "6"
11) "birthdate"
12) "-62467200000000"
13) "address"
14) "923 7 ST NW"
.
.

name: Remove multiple date fields from employee
source:
  db: chinook
  table: employee
transform:
  - uses: remove_field
    with:
      fields:
        - field: hiredate
        - field: birthdate
output:
  - uses: redis.write
    with:
      connection: target
      data_type: hash
      key:
        expression: concat(['emp:', employeeid])
        language: jmespath

> hgetall emp:8
 1) "employeeid"
 2) "8"
 3) "lastname"
 4) "Callahan"
 5) "firstname"
 6) "Laura"
 7) "title"
 8) "IT Staff"
 9) "reportsto"
10) "6"
11) "address"
12) "923 7 ST NW"
.
.


Row Format

Overview

The RDI pipelines support two separate row formats which you can specify in the `source` section of the job file:

- `partial` - (Default) Contains the current value of the row only.
- `full` - Contains all information available for the row, including the key, the before and after values, and the operation code.

The `full` row format is useful when you want to access the metadata associated with the row, such as the operation code, and the before and after values.
The structure of the data passed to the `transform` and `output` sections is different depending on the row format you choose. Consider which row format you are using when you reference keys.
The following two examples demonstrate the difference between the two row formats.

Default row format

With the default row format, the input value is a JSON object containing the current value of the row, and fields can be referenced directly by their name.

Usage example:

[code example]

Full row format {#full}

With `row_format: full` the input value is a JSON object with the following structure:

- `key` - An object containing the attributes of the primary key. For example, `key.id` will give you the value of the `id` column as long as it is part of the primary key.
- `before` - An object containing the previous value of the row.
- `after` - An object containing the current value of the row.
- `opcode` - The operation code. See [Using the operation code]() for more information about the possible opcode values and how to use them.
- `db` - The database name.
- `table` - The table name.
- `schema` - The schema name.

Note: The `db` and `schema` fields are database-specific and may not be available in all databases. For example, MySQL doesn't use `schema` and uses `db` as the database name.

Usage example:

[code example]

Important notes when using `row_format: full`

- The `before` object will be `null` for `insert` and `create` operations, and the `after` object will be `null` for `delete` operations. If you are building the output key manually, you should account for this and ensure that you are not trying to access fields from a `null` object, as shown in the example below:

    [code example]

    Please note that you should not use `key` in combination with `row_format: full` and more than one output, as the `key` object will be overwritten by the previous output. This is a known limitation of the current implementation and is subject to change in future versions.

- The final result of the processing (which is what will be stored in the output) is the value of the `after` object. This means you must reference the fields using the `after` prefix unless you change the output structure in a transformation step. Also, when you add new fields, you must prefix them with `after.` to ensure that they are added to the correct part of the output:

    [code example]

name: Default row format example
source:
  table: addresses
transform:
  - uses: add_field
    with:
      field: city_state
      expression: concat([CITY, ', ', STATE])
      language: jmespath
  - uses: add_field
    with:
      field: op_code_value
      # Operation code is not available in standard row format
      # so the following expression will result in `op_code - None`
      expression: concat(['op_code', ' - ', opcode])
      language: jmespath
output:
  - uses: redis.write
    with:
      data_type: hash
      key:
        expression: concat(['addresses', '#', ID])
        language: jmespath

name: Full row format example
source:
  table: addresses
  row_format: full
transform:
  - uses: add_field
    with:
      # opcode is only available in full row format and can be used in the transformations
      field: after.op_code_value
      expression: address
      language: jmespath
  - uses: add_field
    with:
      field: after.city_state
      # Note that we need to use the `after` prefix to access the current value of the row
      # or `before` to access the previous value
      expression: concat([after.CITY, ', ', after.STATE])
      language: jmespath
output:
  - uses: redis.write
    with:
      data_type: hash
      key:
        # There are different ways to express the key
        # If the `ID` column is the primary key the following expressions
        # are equivalent - `key.ID`, `after.ID`, `values(key)[0]`
        expression: concat(['addresses-full', '#', values(key)[0]])
        language: jmespath

name: Handle delete operations with full row format
    source:
      table: addresses
      row_format: full

    output:
      - uses: redis.write
        with:
          key:
            language: jmespath
            # The following pattern will fail for delete operations. In those cases `after` is null, the resulting key will
            # be 'addresses:None' and the key won't be removed from the target
            # expression: concat(['addresses:', after.ID])

            # This pattern works for all operations, by using the ID from the `after` object if it is available,
            # and falling back to the ID from the `before` object if not.
            expression: concat(['addresses:', after.ID || before.ID])

            # Another option is to use the ID from the `key` object
            # expression: concat(['addresses:', values(key)[0]])

name: Add fields with full row format
    source:
      table: addresses
      row_format: full

    transform:
      - uses: add_field
        with:
          field: after.city_state  # use this to add the new field to the final output
          # field: city_state  # use this if you need a temporary field in the transformation steps, but not in the final output
          expression: concat([after.CITY, ', ', after.STATE])
          language: jmespath


Write to a Redis set

Content

In the example below, data is captured from the source table named `invoice` and is written to a Redis set. The `connection` is an optional parameter that refers to the corresponding connection name defined in `config.yaml`. When you specify the
`data_type` parameter for the job, it overrides the system-wide setting `target_data_type` defined in `config.yaml`.

When writing to a set, you must supply an extra argument, `member`, which specifies the field that will be written. In this case, the result will be a Redis set with key names based on the key expression (for example, `invoices:Germany`, `invoices:USA`) and with an expiration of 100 seconds. If you don't supply an `expire` parameter, the keys will never expire.

[code example]

name: Write invoices to set by country
source:
  schema: public
  table: invoice
output:
  - uses: redis.write
    with:
      connection: target
      data_type: set
      key:
        expression: concat(['invoices:', BillingCountry])
        language: jmespath
      args:
        member: InvoiceId
      expire: 100


Set the key name in the target database

Understanding Default Key Names

When RDI synchronizes data from your primary database to Redis, it automatically generates key names based on a specific pattern.
By default, RDI creates keys using the following format:

* **Single primary key**: `tablename:primarykeyname:primarykeyvalue`
* **Composite primary keys**: `tablename:key1name:key1value:key2name:key2value`

Examples

* For a table named `employee` with primary key `employeeid`, a record with `employeeid=1` will have the key `employee:employeeid:1`
* For a table named `orders` with composite primary keys `orderid` and `customerid`, a record with `orderid=1` and `customerid=2` will have the key `orders:orderid:1:customerid:2`

Customizing Key Names

While the default key naming convention works for many use cases, you may need custom key formats to:

* Match existing Redis key patterns
* Create more concise key names
* Implement application-specific naming schemes
* Optimize for specific access patterns

To customize key names, use the `key` section within the `redis.write` output configuration. This section requires two parameters:

* `expression`: Defines the custom key format using a supported expression language
* `language`: Specifies the expression language to use (`jmespath` or `sql`)

Example

[code example]

Special Considerations for Full Row Format

When working with the full row format, you need to handle key generation differently to ensure proper behavior across all operation types (create, update, and delete). You must reference attributes correctly to ensure consistent key generation, especially for delete operations where the "after" state is empty. The example below demonstrates how to handle this:

[code example]

Summary

Proper key naming is essential for effective data organization in Redis. RDI provides:

1. **Default key naming** that follows a consistent pattern based on table and primary key information
2. **Custom key naming** through the `key` section with `expression` and `language` parameters
3. **Special handling for full row format** to ensure consistent key generation across all operation types

By understanding and utilizing these options, you can ensure your Redis keys are optimally structured for your specific use case, making data retrieval more efficient and your Redis implementation more maintainable.

name: Custom key name for customers
source:
  db: inventory
  table: customers
output:
  - uses: redis.write
    with:
      key:
        expression: concat(['customers', '#', id])
        language: jmespath

name: Custom key with full row format
source:
  db: inventory
  table: customers
  row_format: full
output:
  - uses: redis.write
    with:
      key:
        # Here we use the operation code to determine the value of the key to ensure that
        # delete operations will result in the correct key being deleted
        expression: concat(['customers', '#', opcode == 'd' && before.id || after.id])
        language: jmespath


Write to a Redis sorted set

Content

In the example below, data is captured from the source table named `invoice` and is written to a Redis sorted set. The `connection` is an optional parameter that refers to the corresponding connection name defined in `config.yaml`. When
you specify the `data_type` parameter for the job, it overrides the system-wide setting `target_data_type` defined in `config.yaml`.

When writing to sorted sets, you must provide two additional arguments, `member` and `score`. These specify the field names that will be used as a member and a score to add an element to a sorted set. In this case, the result will be a Redis sorted set named `invoices:sorted` based on the key expression and with an expiration of 100 seconds for each set member. If you don't supply an `expire` parameter, the keys will never expire.

[code example]

Since sorted sets in Redis are inherently sorted, you can easily get the top N invoices by total invoice amount using the command below (the range 0..9 gets the top 10 invoices):

[code example]

name: Write invoices to sorted set by total
source:
  schema: public
  table: invoice
output:
  - uses: redis.write
    with:
      connection: target
      data_type: sorted_set
      key:
        expression: "`invoices:sorted`"
        language: jmespath
      args:
        score: Total
        member: InvoiceId
      expire: 100

ZREVRANGE invoices:sorted 0 9 WITHSCORES


Using SQL CASE

Overview

The [`CASE`](https://www.w3schools.com/sql/sql_case.asp) statement allows you to specify conditions and return different values based on those conditions. You can use it both to create new fields or filter existing data.

Using SQL CASE to create a new field

The example below demonstrates how to use the `CASE` statement to create a new field called `Market` based on the value of the `BillingCountry` field in the `Invoice` table. The new field categorizes countries into regions such as "North America" and "Europe".

[code example]

Using SQL CASE to filter data

You can also use the `CASE` statement to filter data based on specific conditions. The example below demonstrates how to filter the `Invoice` table to include only invoices from the USA and Canada that have a `Total` value above their country-specific threshold.

Because the `Total` field is a Decimal in the source table, it is represented as a string in Debezium and so you must cast it to `REAL` to compare it numerically in the `CASE` statement. Without this cast, it will be compared as a string value, which will give the wrong result.

[code example]

name: Add market field using SQL CASE
source:
  table: Invoice

transform:
  - uses: add_field
    with:
      field: "Market"
      language: sql
      expression: |
        CASE
          WHEN BillingCountry = 'USA' THEN 'North America'
          WHEN BillingCountry = 'Canada' THEN 'North America'
          WHEN BillingCountry = 'UK' THEN 'Europe'
          WHEN BillingCountry = 'France' THEN 'Europe'
          ELSE 'Other'
        END

name: Filter invoices by country and total
source:
  table: Invoice

transform:
  - uses: filter
    with:
      language: sql
      expression: |
        CASE
          WHEN BillingCountry = 'USA' AND CAST(Total AS REAL) > 11.99 THEN True
          WHEN BillingCountry = 'Canada' AND CAST(Total AS REAL) > 9.99 THEN True
          ELSE False
        END


Write to a Redis stream

Content

In the example below, data is captured from the source table named `invoice` and is written to a Redis stream. The `connection` is an optional parameter that refers to the corresponding connection name defined in `config.yaml`.
When you specify the `data_type` parameter for the job, it overrides the system-wide setting `target_data_type` defined in `config.yaml`.

When writing to streams, you can use the optional parameter `mapping` to limit the number of fields sent in a message and to provide aliases for them. If you don't use the `mapping` parameter, all fields captured in the source will be passed as the message payload.

Note that streams are different from other data structures because existing messages are never updated or deleted. Any operation in the source will generate a new message with the corresponding operation code (`op_code` field) that is automatically added to the message payload.

In this case, the result will be a Redis stream with the name based on the key expression (for example, `invoice:events`) and with an expiration of 100 seconds for the whole stream. If you don't supply an `expire` parameter, the keys will never expire.

In the example, only three original fields are passed in the message payload: `InvoiceId` (as `message_id`), `BillingCountry` (as `country`), `Total` (as `Total`, no alias provided) and `op_code`, which is implicitly added to all messages sent to streams.

[code example]

name: Write invoice events to stream
source:
  schema: public
  table: invoice
output:
  - uses: redis.write
    with:
      connection: target
      data_type: stream
      key:
        expression: "`invoice:events`"
        language: jmespath
      mapping:
        - InvoiceId: message_id
        - BillingCountry: country
        - Total
      expire: 100


Write to a Redis string

Content

The string data type is useful for capturing a string representation of a single column from
a source table.

In the example job below, the `title` column is captured from the `album` table in the source.
The `title` is then written to the Redis target database as a string under a custom key of the
form `AlbumTitle:42`, where the `42` is the primary key value of the table (the `albumid` column).

The `connection` is an optional parameter that refers to the corresponding connection name defined in
[`config.yaml`]().
When you specify the `data_type` parameter for the job, it overrides the system-wide setting `target_data_type` defined in `config.yaml`. Here, the `string` data type also requires an `args` subsection
with a `value` argument that specifies the column you want to capture from the source table.

The optional `expire` parameter sets the length of time, in seconds, that a new key will
persist for after it is created (here, it is 86400 seconds, which equals one day).
After this time, the key will be deleted automatically.
If you don't supply an `expire` parameter, the keys will never expire.

[code example]

name: Write album title to string
source:
  table: album
  row_format: full
output:
  - uses: redis.write
    with:
      connection: target
      data_type: string
      key:
        expression: concat(['AlbumTitle:', values(key)[0]])
        language: jmespath
      args:
        value: title
      expire: 86400


Writing to multiple keys

Content

If you want to write results to multiple keys, you can do so by defining multiple `redis.write` subsections in the `output` section of the job file. Each instance of `redis.write` can specify a different key, data format, and other parameters. For example, you can create two different keys for the same data, one with a default key format and another with a custom key format and mapping.

[code example]

name: Write to multiple keys with different formats
output:
  - uses: redis.write
    with:
      # Setting data_type to JSON and using the default key format
      data_type: json

  - uses: redis.write
    with:
      data_type: json

      # Defining a custom key format
      key:
        language: jmespath
        expression: concat(['events-simplified:id:', id])

      # And defining a custom mapping
      mapping:
        - id: id
        - name: name
        - location: location


Remapping the fields in the output

Content

Sometimes, you may want to remap the fields in the output of a data pipeline. You can do this by defining a `mapping` section in the output configuration.

[code example]

The example above remaps the `CustomerId` field to `id`, `FirstName` to `first_name`, and `LastName` to `last_name` in the output. This allows you to customize the field names in the Redis data store according to your application's requirements.
You can also use `mapping` to include only the fields you need in the output and exclude the rest.

Mapping only allows you to rename fields and limit the output to specific fields and define a single level structure. To create nested structures and/or perform operations on the field values you can use the [map transformation]().

name: Remap customer fields
source:
  table: Customer

output:
  - uses: redis.write
    with:
      data_type: hash
      mapping:
        - CustomerId: id
        - FirstName: first_name
        - LastName: last_name


FAQ

Summary: Find answers to common questions about RDI

Which license does RDI use?

You must purchase a commercial license for RDI with Redis Enterprise. This includes two extra
Redis Enterprise shards (primary and replica) for the staging database.

How does RDI track data changes in the source database?

RDI uses mechanisms that are specific for each of the supported
source databases:

- **Oracle**:  RDI uses `logminer` to parse the Oracle `binary log` and `archive logs`. This
  lists any changes in a database view that RDI can query.
- **MySQL/MariaDB**: RDI uses `binary log replication` to get all the commits.
- **PostgreSQL**:  RDI uses the `pgoutput` plugin.
- **SQL Server**: RDI uses the CDC mechanism.

How much data can RDI process?

RDI uses the concept of *processing units*. Each processing unit uses 1 CPU core and can process
about 10,000 records per second, assuming the records have a size of about 1KB each. This throughput
might change slightly depending on the number of columns, the number of data transformations,
and the speed of the network. Typically, one processing unit is enough for RDI to deal with the
traffic from a relational database.

Can RDI work with any Redis database?

No. RDI is designed and tested to work only with Redis Enterprise. The staging database can
only use version 6.4 or above. The target Redis database can be of any version and can be a
replica of an Active-Active replication setup or an Auto tiering database.

Can RDI automatically track changes to the source database schema?

If you don't configure RDI to capture a specific set of tables in the schema then it will
detect any new tables when they are added. Similarly, RDI will capture new table columns
and changes to column names unless you configure it for a specific set of columns.
Bear in mind that the Redis keys in the target database will change to reflect the
new or renamed tables and columns.

Should I be concerned when the log says RDI is out of memory? {#rdi-oom}

Sometimes the Debezium log will contain a message saying that RDI is out of
memory. This is not an error but an informative message to say that RDI
is applying *backpressure* to Debezium. See
[Backpressure mechanism]()
in the Architecture guide for more information.

What happens when RDI can't write to the target Redis database?

RDI will keep attempting to write the changes to the target and will also attempt
to reconnect to it, if necessary. While the target is disconnected, RDI
will keep capturing change events from the source database and adding them to its
streams in the staging database. This continues until the staging database gets
low on space to store new events. When RDI detects this, it applies a "back pressure"
mechanism to capture data from the source less frequently, which reduces the risk of running
out of space altogether. The systems that the source databases use to record changes can
retain the change data for at least a few hours, and RDI can catch up with the
changes as soon as the target connection recovers or the staging database has
more space available.

What does RDI do if the data is corrupted or invalid?

The collector reports the data to RDI in a structured JSON format. If
the structure of the JSON data is invalid or if there is a fatal bug in the transformation
job then RDI can't transform the data. When this happens, RDI will store the original data
in a "dead letter queue" along with a message to say why it was rejected. The dead letter
queue is stored as a capped stream in the RDI staging database. You can see its contents
with Redis Insight or with the
[`redis-di get-rejected`]()
command from the CLI.

Can I use RDI without persistence enabled?

By default, RDI requires persistence to be enabled on the RDI database. This ensures that RDI can recover both its configuration and the last known state if the cluster crashes.

If you don't have permissions to use persistence due to compliance or other reasons, you can disable
the persistence check on the RDI database (Helm installation only). If you do this, RDI will not be
able to recover from a crash, and you will have to perform a new deploy to reinitialize the pipeline.

To disable the persistence check, set the `aofRequired` value to `false` in the `operator->prerequisiteChecks`
section of the `values.yaml` file.

This option is available in RDI 1.16.2 and later.


Redis Data Integration

Summary: Understand when (and when not) to use RDI.

When to use RDI

Summary: Get started with a simple pipeline example

Quickstart

Summary: Learn how to install and upgrade RDI

Install and upgrade

Summary: Discover the main components of RDI

Architecture

Summary: Learn how to configure RDI for data capture and transformation.

Data pipelines

Summary: Learn how to monitor RDI

Observability

FAQ

Summary: Solve and report simple problems with RDI

Troubleshooting

Summary: View reference material for Redis Data Integration

Reference

Release notes

Summary: Describes where to view the preview version for RDI products

Preview version

Overview

This is the first General Availability version of Redis Data Integration (RDI).

RDI's purpose is to help Redis customers sync Redis Enterprise with live data from their slow disk based databases in order to:

- Meet the required speed and scale of read queries and provide an excellent and predictable user experience.
- Save resources and time when building pipelines and coding data transformations.
- Reduce the total cost of ownership by saving money on expensive database read replicas.

If you use a relational database as the system of record for your app,
you may eventually find
that its performance doesn't scale well as your userbase grows. It may be
acceptable for a few thousand users but for a few million, it can become a
major problem. If you don't have the option of abandoning the relational
database, you should consider using a fast
database, such as Redis, to cache data from read queries. Since read queries
are typically many times more common than writes, the cache will greatly
improve performance and let your app scale without a major redesign.

RDI keeps a Redis cache up to date with changes in the primary database, using a
[*Change Data Capture (CDC)*](https://en.wikipedia.org/wiki/Change_data_capture) mechanism.
It also lets you *transform* the data from relational tables into convenient
and fast data structures that match your app's requirements. You specify the
transformations using a configuration system, so no coding is necessary.

RDI is supported with Redis databases or [CRDB](https://redis.io/active-active/) (Active Active Replication) targets, and is also available on [Redis Cloud]().

Features

RDI provides enterprise-grade streaming data pipelines with the following features:

- **Near realtime pipeline** - The CDC system captures changes in very short time intervals,
  then ships and processes them in *micro-batches* to provide near real time updates to Redis.
- **At least once guarantee** - RDI will deliver any change to the selected data set at least
  once to the target Redis database.
- **Data integrity** - RDI keeps the data change order per source table or unique key.
- **High availability** - All stateless components have hot failover or quick automatic recovery.
  RDI state is always highly available using Redis Enterprise replication.
- **Easy to install and operate** - Use a self-documenting command line interface (CLI)
  for all installation and day-two operations.
- **No coding needed** - Create and test your pipelines using Redis Insight.
- **Data-in-transit encryption** - RDI never persists data to disk. All data in-flight is
  protected using TLS or mTLS connections.
- **Observability - Metrics** - RDI collects data processing counters at source table granularity
  along with data processing performance metrics. These are available via GUI, CLI and
  [Prometheus](https://prometheus.io/) endpoints.
- **Observability - logs** - RDI saves rotating logs to a single folder. They are in a JSON format,
  so you can collect and process them with your favorite observability tool.
- **Backpressure mechanism** - RDI is designed to backoff writing data when the cache gets
  disconnected, which prevents cascading failure. Since the change data is persisted in the source
  database and Redis is very fast, RDI can easily catch up with missed changes after a short period of
  disconnection. See [Backpressure mechanism]() for more information.
- **Recovering from full failure** - If the cache fails or gets disconnected for a long time,
  RDI can reconstruct the cache data in Redis using a full snapshot of the defined dataset.
- **High throughput** - Because RDI uses Redis for staging and writes to Redis as a target,
  it has very high throughput. With a single processor core and records of about 1KB in size,
  RDI processes around 10,000 records per second. While taking the initial full *snapshot* of
  the source database, RDI automatically scales to a configurable number of processing units,
  to fill the cache as fast as possible.

When to use RDI

RDI is highly configurable but it is not intended to be a general
solution for all data integration tasks. See
[When to use RDI]()
to find out if your use case is a good fit for RDI's features.

Supported source databases

RDI can capture data from any of the following sources:

Continue learning with Redis University

* [Redis Data Integration Lab](https://university.redis.io/course/2qa1u1ss21vsy5?tab=details)

Documentation

Learn more about RDI from the other pages in this section:


Test HA failover

Summary: Learn how to perform HA failover testing for Redis Data Integration (RDI) to ensure high availability and reliability of your data integration setup.

Setup

1. Ensure that RDI is up and running on both primary and secondary nodes.
   Run the following command and verify and that each instance should show healthy and running `rdi-api` and `rdi-operator` pods.
[code example]

2. Identify the leader node - this is the one that has a running `collector-source` pod.

Performing the HA Failover Testing

To perform HA, you can simulate a connection failure between the leader and the RDI database by blocking the network traffic. You can do this by running the following commands on the leader node:

1. Identify the RDI database IP (replace `<hostname>` with your own hostname):
[code example]

2. For each of the IPs returned by the above command, run the following command to block the traffic:

[code example]

The default configuration for the leader lock is 60 seconds, so it may take up to 2 minutes for the failover to occur.
Meanwhile you can follow the logs of the operator to see the failover process:

[code example]

In about 10 seconds you will start seeing log entries from the leader saying that it could not acquire the leadership.
When the leader lock expires, the second node will acquire the leadership and you will see log entries from the second node indicating that it has become the leader.

Cleanup

To clean up after the test, remove the `iptables` rule that you added to block the traffic:

```sudo iptables -D FORWARD -d <databse_ip> -j DROP```

Use `sudo iptables -S | grep <database_ip>` to verify that the rule has been removed.

kubectl -n rdi get pods

# Example output:
NAME                                   READY   STATUS      RESTARTS      AGE
collector-api-577d95bfd8-5wbg6         1/1     Running     0             12m
collector-source-95f45bcf7-vwn5l       1/1     Running     0             12m
fluentd-zq2lc                          1/1     Running     0             72m
logrotate-29530445-j729x               0/1     Completed   0             14m
logrotate-29530450-dprr2               0/1     Completed   0             9m40s
logrotate-29530455-mfmzw               0/1     Completed   0             4m40s
processor-f66655469-h7nw2              1/1     Running     0             12m
rdi-api-f75df6796-qwqjw                1/1     Running     0             72m
rdi-metrics-exporter-d57cdf8c8-wjzb5   1/1     Running     0             72m
rdi-operator-7f7f6c7dfd-5qmjd          1/1     Running     0             71m
rdi-reloader-77df5f7854-lwmvz          1/1     Running     0             71m

dig +short <hostname>

# Example:
# dig +short my.redis.hostname.com

# Example output:
54.78.220.161

sudo iptables -I FORWARD -d <database_ip> -j DROP

# With the IP from the example above, the command would be:
sudo iptables -I FORWARD -d 54.78.220.161 -j DROP

kubectl -n rdi logs rdi-operator-7f7f6c7dfd-5qmjd -f


Install and upgrade

Summary: Learn how to install and upgrade RDI

Summary: Requirements and recommendations for RDI installations.

Requirements summary

Summary: Learn how to install RDI on one or more VMs

Install on VMs

Summary: Learn how to install RDI on Kubernetes

Install on Kubernetes

Summary: Learn how to upgrade an existing RDI installation

Upgrading RDI

Test HA failover

Content

The guides in this section explain the options you have for installing and upgrading RDI on your own servers. See the [Redis Cloud RDI guide]() to
learn how to set up RDI for a cloud database.
Before you use RDI, you must also configure your source database to enable CDC. See the
[Prepare source databases]()
section to learn how to do this.


Install on Kubernetes

Summary: Learn how to install RDI on Kubernetes

Overview

This guide explains how to use the RDI [Helm chart](https://helm.sh/docs/topics/charts/)
to install on [Kubernetes](https://kubernetes.io/) (K8s). You can also
[Install RDI on VMs]().

The installation creates the following K8s objects:

-   A K8s [namespace](https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/) named `rdi`.
    You can also use a different namespace name if you prefer.
-   [Deployments](https://kubernetes.io/docs/concepts/workloads/controllers/deployment/) and
    [services](https://kubernetes.io/docs/concepts/services-networking/service/) for the
    [RDI operator](),
    [metrics exporter](), and API server.
-   A [service account](https://kubernetes.io/docs/concepts/security/service-accounts/)
    and [RBAC resources](https://kubernetes.io/docs/reference/access-authn-authz/rbac) for the RDI operator.
-   A [ConfigMap](https://kubernetes.io/docs/concepts/configuration/configmap/) with RDI database details.
-   [Secrets](https://kubernetes.io/docs/concepts/configuration/secret/)
    with the RDI database credentials and TLS certificates.
-   Other optional K8s resources such as [ingresses](https://kubernetes.io/docs/concepts/services-networking/ingress/)
    that can be enabled depending on your K8s environment and needs.

You can use this installation on [OpenShift](https://docs.openshift.com/) and other K8s distributions
including cloud providers' K8s managed clusters.

You can configure the RDI Helm chart to pull the RDI images from [dockerhub](https://hub.docker.com/u/redis)
or from your own [private image registry](#using-a-private-image-registry).

Before you install

Complete the following steps before installing the RDI Helm chart:

-   [Create the RDI database](#create-the-rdi-database) on your Redis Enterprise cluster.

-   Create a [user]()
    for the RDI database if you prefer not to use the default password (see
    [Access control]() for
    more information).

-   Download the RDI Helm chart tar file from the
    [Redis download center](https://redis-enterprise-software-downloads.s3.amazonaws.com/redis-di/rdi-.tgz) (in the *Modules, Tools & Integration* category) .

    [code example]

-   If you want to use a private image registry,
    [prepare it with the RDI images](#using-a-private-image-registry).

Create the RDI database

RDI uses a database on your Redis Enterprise cluster to store its state
information. Use the Redis Enterprise Cluster Manager UI to create the RDI database with the following
requirements:

You should then provide the details of this database in the [`values.yaml`](#the-valuesyaml-file)
file as described below.

Using a private image registry

Add the RDI images from [dockerhub](https://hub.docker.com/u/redis) to your local registry.
You need the following RDI images with tags matching the RDI version you want to install:

-   [redis/rdi-api](https://hub.docker.com/r/redis/rdi-api)
-   [redis/rdi-operator](https://hub.docker.com/r/redis/rdi-operator)
-   [redis/rdi-monitor](https://hub.docker.com/r/redis/rdi-monitor)
-   [redis/rdi-processor](https://hub.docker.com/r/redis/rdi-processor)
-   [redis/rdi-collector-api](https://hub.docker.com/r/redis/rdi-collector-api)
-   [redis/rdi-collector-initializer](https://hub.docker.com/r/redis/rdi-collector-initializer)

If you plan to use Spanner as a source for your pipeline, you’ll need an additional image.: [redis/rdi-flink-collector](https://hub.docker.com/r/redis/rdi-flink-collector).

In addition, the RDI Helm chart uses the following 3rd party images:

-   [redislabs/debezium-server:3.0.8.Final-rdi.1](https://hub.docker.com/r/redislabs/debezium-server),
    based on `quay.io/debezium/server/3.0.8.Final` with minor modifications:
    [Debezium](https://debezium.io/), an open source distributed platform for change data capture.
-   [redis/reloader:v1.1.0](https://hub.docker.com/r/redis/reloader), originally `ghcr.io/stakater/reloader:v1.1.0`:
    [Reloader](https://github.com/stakater/Reloader), a K8s controller to watch changes to ConfigMaps
    and Secrets and do rolling upgrades.
-   [redis/kube-webhook-certgen:v20221220-controller-v1.5.1-58-g787ea74b6](https://hub.docker.com/r/redis/kube-webhook-certgen),
    originally `registry.k8s.io/ingress-nginx/kube-webhook-certgen/v20221220-controller-v1.5.1-58-g787ea74b6`:
    [kube-webhook-certgen](https://github.com/jet/kube-webhook-certgen), K8s webhook certificate generator and patcher.

The example below shows how to specify the registry and image pull secret in your
[`rdi-values.yaml`](#the-valuesyaml-file) file for the Helm chart:

[code example]

To pull images from a private image registry, you must provide the image pull secret and in some cases also set the permissions. Follow the links below to learn how to use a private registry with:

-   [Rancher](https://ranchermanager.docs.rancher.com/how-to-guides/new-user-guides/kubernetes-resources-setup/kubernetes-and-docker-registries#using-a-private-registry)
-   [OpenShift](https://docs.openshift.com/container-platform/4.17/openshift_images/managing_images/using-image-pull-secrets.html)
-   [Amazon Elastic Kubernetes Service (EKS)](https://docs.aws.amazon.com/AmazonECR/latest/userguide/ECR_on_EKS.html)
-   [Google Kubernetes Engine (GKE)](https://cloud.google.com/artifact-registry/docs/pull-cached-dockerhub-images)
-   [Azure Kubernetes Service (AKS)](https://learn.microsoft.com/en-us/azure/aks/cluster-container-registry-integration?tabs=azure-cli)

Supported versions of Kubernetes and OpenShift

Install the RDI Helm chart

1.  Scaffold the default `values.yaml` file from the chart into a local
    `rdi-values.yaml` file:

    [code example]

1.  Open the `rdi-values.yaml` file you just created, change or add the appropriate
    values for your installation, and delete the values you have not changed to
    use their default values.
    See [The `values.yaml` file](#the-valuesyaml-file) for more details.

1.  Run the `helm upgrade --install` command:

    [code example]

    The above command will install RDI in a namespace called
    `rdi`. If you want to use a different namespace, pass the option
    `-n <custom-namespace>` to the `helm install` command instead.

The `values.yaml` file

The [`values.yaml`](https://helm.sh/docs/topics/charts/#templates-and-values) file inside the
Helm chart contains the values you can set for the RDI Helm installation.
See the comments by each value for more information about the values you may need to add or change
depending on your use case.

At a minimum, you must set the values of `connection.host`, `connection.port`, and `connection.password`
to enable the basic connection to the RDI database.
You must also set `api.jwtKey`, RDI uses this value to encrypt the
[JSON web token (JWT)](https://jwt.io/) token used by RDI API. Best practice is
to generate a value containing 32 random bytes of data (equivalent to 256
bits) and then encode this value as ASCII characters. Use the following
command to generate the random key from the
[`urandom` special file](https://en.wikipedia.org/wiki//dev/random):

[code example]

If you use TLS to connect to the RDI database, you must set the
CA certificate content in `connection.ssl.cacert` (for TLS). In addition, if you
also use mTLS, you must set the client certificate and private key contents in
`connection.ssl.cert`, and `connection.ssl.key`.

-   You can add the certificate content directly in the `rdi-values.yaml` file
    as follows:

    [code example]

-   Alternatively, you can use the `--set-file` argument to set these values to
    the content of your certificate files as follows:

    [code example]

Please see [these docs]() if this RDI installation is for use with GCP Spanner.

If you are deploying to [OpenShift](https://docs.openshift.com/), you must
set `global.openshift` to `true`:

[code example]

Set `global.securityContext.runAsUser` and
`global.securityContext.runAsGroup` to the appropriate values for your
OpenShift environment.

[code example]

The default OpenShift Security Context Constraints (SCCs)
will not allow RDI to run if `global.securityContext.runAsUser`
and `global.securityContext.runAsGroup` have their default values of `1000`.
You must edit your `rdi-values.yaml` file to ensure these values are
in the valid range for your OpenShift environment.

Use the following [OpenShift CLI](https://docs.redhat.com/en/documentation/openshift_container_platform/4.19/html/cli_tools/openshift-cli-oc) command
to find the user and group ranges for your project:

[code example]

Check the installation

To verify the status of the K8s deployment, run the following command:

[code example]

The output looks like the following. Check that the `rdi` release is listed.
With RDI 1.8.0 or later, check that the `default` release is also listed.

[code example]

Also, check that all pods have `Running` status:

[code example]

You can verify that the RDI API works by adding a connection to the RDI API server to
[Redis Insight]().

Using ingress controllers

You must ensure that an appropriate
[ingress controller](https://kubernetes.io/docs/concepts/services-networking/ingress-controllers/)
is available in your K8s cluster to expose the RDI API service via the K8s
[`Ingress`](https://kubernetes.io/docs/concepts/services-networking/ingress/)
resource. Follow the documentation of your cloud provider or of
the ingress controller to install the controller correctly.

Using the `nginx` ingress controller on AKS

On AKS, if you want to use the open source
[`nginx`](https://nginx.org/)
[ingress controller](https://github.com/kubernetes/ingress-nginx/blob/main/README.md#readme)
rather than the
[AKS application routing add-on](https://learn.microsoft.com/en-us/azure/aks/app-routing),
follow the AKS documentation for
[creating an unmanaged ingress controller](https://learn.microsoft.com/en-us/troubleshoot/azure/azure-kubernetes/load-bal-ingress-c/create-unmanaged-ingress-controller?tabs=azure-cli).
Specifically, ensure that one or both of the following Helm chart values is set:

- `controller.service.annotations."service\.beta\.kubernetes\.io/azure-load-balancer-health-probe-request-path"=/healthz`
- `controller.service.externalTrafficPolicy=Local`

Prepare your source database

Before deploying a pipeline, you must configure your source database to enable CDC. See the
[Prepare source databases]()
section to learn how to do this.

Deploy a pipeline

When the Helm installation is complete and you have prepared the source database for CDC,
you are ready to start using RDI.
Use [Redis Insight]() to
[configure]() and
[deploy]()
your pipeline (see
[RDI in Redis Insight]()
for full details on how to do this).

Uninstall RDI

If you want to remove your RDI K8s installation, first run
the following commands. (If you installed RDI into a custom namespace then
replace `rdi` with the name of your namespace.)

[code example]

The line `kubectl delete pipeline default -n rdi` is only needed for RDI 1.8.0 or above.

If you also want to delete the keys from your RDI database, connect to it with
[`redis-cli`]() and run a
[`FLUSHALL`]() command.

export RDI_VERSION=
    wget https://redis-enterprise-software-downloads.s3.amazonaws.com/redis-di/rdi-$RDI_VERSION.tgz

global:
  # Global image settings.
  # If using a private image registry, update the default values accordingly.
  image:
    registry: your-registry
    repository: your-repository # If different from "redis"

  # Image pull secrets to be used when using a private image registry.
  imagePullSecrets:
    - name: your-secret-name

# ...

# Configuration of the reloader.
reloader:
  reloader:
    # ...
    deployment:
      image:
        name: my-registry.com/my-repo/reloader
        #...

helm show values rdi-<tag>.tgz > rdi-values.yaml

helm upgrade --install rdi rdi-<tag>.tgz -f rdi-values.yaml -n rdi --create-namespace

helm upgrade --install rdi rdi-<tag>.tar.gz -f rdi-values.yaml -n rdi --create-namespace \
      --set connection.ssl.enabled=true \
      --set-file connection.ssl.cacert=<path-to-CA-certificate> \
      --set-file connection.ssl.cert=<path-to-client-certificate> \
      --set-file connection.ssl.key=<path-to-client-key>

global:
  # Indicates whether the deployment is intended for an OpenShift environment.
  openShift: true

global:
  # Container default security context.
  # ref: https://kubernetes.io/docs/tasks/configure-pod-container/security-context/#set-the-security-context-for-a-container
  securityContext:
    runAsNonRoot: true
    # On OpenShift, user and group 1000 are usually not allowed.
    # If using OpenShift, set runAsUser and runAsGroup to values in your project's user and group ranges.
    # You can examine the latter via `oc get projects <rid-project-name> -o yaml | grep "openshift.io/sa.scc"`
    runAsUser: 1000701234
    runAsGroup: 1000701234
    allowPrivilegeEscalation: false

NAME   	NAMESPACE	REVISION	UPDATED         STATUS  	CHART         	APP VERSION
default	rdi      	1       	2025-05-08 ... 	deployed	pipeline-0.1.0	<tag>
rdi    	rdi      	3       	2025-05-08 ...	deployed	rdi-1.0.0

kubectl get pod -n rdi

NAME                      READY  STATUS  	RESTARTS  AGE
collector-api-<id>        1/1    Running  0         29m
rdi-api-<id>              1/1 	 Running 	0      	  29m
rdi-metric-exporter-<id>  1/1    Running 	0      	  29m
rdi-operator-<id>         1/1 	 Running 	0      	  29m
rdi-reloader-<id>         1/1 	 Running 	0      	  29m

kubectl delete pipeline default -n rdi
helm uninstall rdi -n rdi
kubectl delete namespace rdi


Install on VMs

Overview

This guide explains how to install Redis Data Integration (RDI) on one or more VMs and integrate it with
your source database. You can also
[Install RDI on Kubernetes]().

We recommend you always use the latest version, which is RDI v.

Create the RDI database

RDI uses a database on your Redis Enterprise cluster to store its state
information. Use the Redis Enterprise Cluster Manager UI to create the RDI database with the following
requirements:

Hardware sizing

RDI is mainly CPU and network bound.
Each of the RDI VMs should have at least:

VM Installation Requirements

You would normally install RDI on two VMs for High Availability (HA) but you can also install
just one VM if you don't need this. For example, you might not need HA during
development and testing.

You can't install RDI on a host where a Redis Enterprise cluster
is also installed, due to incompatible network rules. If you want to install RDI on a
host that you have previously used for Redis Enterprise then you must
use [`iptables`](https://www.netfilter.org/projects/iptables/index.html) to
"clean" the host before installation with the following command line:

[code example]

You may encounter problems if you use `iptables` v1.6.1 and earlier in
`nftables` mode. Use `iptables` versions later than v1.6.1 or enable the `iptables`
legacy mode with the following commands:

[code example]

Also, `iptables` versions 1.8.0-1.8.4 have known issues that can prevent RDI
from working, especially on RHEL 8. Ideally, use `iptables` v1.8.8, which is
known to work correctly with RDI.

The supported OS versions for RDI are:

You must run the RDI installer as a privileged user because it installs
[containerd](https://containerd.io/) and registers services. However, you don't
need any special privileges to run RDI processes for normal operation.

RDI has a few
requirements for cloud VMs that you must implement before running the
RDI installer, or else installation will fail. The following sections
give full pre-installation instructions for [RHEL](#firewall-rhel) and
[Ubuntu](#firewall-ubuntu).

RHEL {#firewall-rhel}

We recommend you turn off
[`firewalld`](https://firewalld.org/documentation/)
before installation using the command:

[code example]

However, if you do need to use `firewalld`, you must add the following rules:

[code example]

If you have `nm-cloud-setup.service` enabled, you must disable it and reboot the
node with the following commands:

[code example]

Ubuntu {#firewall-ubuntu}

We recommend you turn off
[Uncomplicated Firewall](https://wiki.ubuntu.com/UncomplicatedFirewall) (`ufw`)
before installation with the command:

[code example]

However, if you do need to use `ufw`, you must add the following rules:

[code example]

Installation steps

Follow the steps below for each of your VMs.

RDI installs executables by default in the `/var` partition, so you must
ensure it is mounted without the `noexec` option. Use the following command to
find any partitions mounted with the `noexec` option:

[code example]

If your `/var` partition is listed in the output from this command, you must remount
it without the `noexec` option. See
[Using the mount command](https://docs.redhat.com/en/documentation/red_hat_enterprise_linux/5/html/deployment_guide/chap-using_the_mount_command) in the Red Hat documentation to learn how to remount a partition.

1.  Download the RDI installer from the
    [Redis download center](https://redis-enterprise-software-downloads.s3.amazonaws.com/redis-di/rdi-installation-.tar.gz)
    (from the *Modules, Tools & Integration* category) and extract it to your preferred installation
    folder.

    [code example]

1.  Go to the installation folder:

    [code example]

1.  Run the `install.sh` script as a privileged user:

    [code example]

    RDI uses [K3s](https://k3s.io/) as part of its implementation.
    By default, the installer installs K3s in the `/var/lib` directory,
    but this might be a problem if you have limited space in `/var`
    or your company policy forbids you to install there. You can
    select a different directory for the K3s installation using the
    `--installation-dir` option with `install.sh`:
[code example]

    **Advanced**: You can also pass custom K3s parameters to the installer using the
    `INSTALL_K3S_EXEC` environment variable. For example, to set the kubeconfig file
    permissions to be readable by all users:

    [code example]

    You can combine multiple K3s options in the `INSTALL_K3S_EXEC` variable. See the
    [K3s documentation](https://docs.k3s.io/installation/configuration) for a full list of
    available options.

    Only modify K3s parameters if you understand exactly what you are changing
    and why. Incorrect K3s configuration can cause RDI installation to fail or result in an
    unstable deployment.

The RDI installer collects all necessary configuration details and alerts you to potential issues,
offering options to abort, apply fixes, or provide additional information.
Once complete, it guides you through creating secrets and setting up your pipeline.

It is strongly recommended to specify a hostname rather than an IP address for
connecting to your RDI database, for the following reasons:

-   Any DNS resolution issues will be detected during the installation rather than
    later during pipeline deployment.
-   If you use TLS, your RDI database CA certificate must contain the hostname you specified
    either as a common name (CN) or as a subject alternative name (SAN). CA certificates
    usually don't contain IP addresses.

If you specify `localhost` as the address of the RDI database server during
installation then the connection will fail if the actual IP address changes for the local
VM. For this reason, we recommend that you don't use `localhost` for the address. However,
if you do encounter this problem, you can fix it using the following commands on the VM
that is running RDI itself:

[code example]

After the installation is finished, RDI is ready for use.

Supply cloud DNS information

This section is only relevant if you are installing RDI
on VMs in a cloud environment.

If you are using [Amazon Route 53](https://aws.amazon.com/route53/),
[Google Cloud DNS](https://cloud.google.com/dns?hl=en), or
[Azure DNS](https://azure.microsoft.com/en-gb/products/dns)
then you must supply the installer with the nameserver IP address
during installation. The table below
shows the appropriate IP address for each cloud provider:

| Platform | Nameserver IP |
| :-- | :-- |
| [Amazon Route 53](https://aws.amazon.com/route53/) | 169.254.169.253 |
| [Google Cloud DNS](https://cloud.google.com/dns?hl=en) | 169.254.169.254 |
| [Azure DNS](https://azure.microsoft.com/en-gb/products/dns) | 168.63.129.16 |

If you are using Route 53, you should first check that your VPC
is configured to allow it. See
[DNS attributes in your VPC](https://docs.aws.amazon.com/vpc/latest/userguide/AmazonDNS-concepts.html#vpc-dns-support)
in the Amazon docs for more information.

Installing with High Availability

To install RDI with High Availability (HA), perform the [Installation steps](#installation-steps)
on two different VMs. The first VM will automatically become the active (primary) instance,
while the second VM will become the passive (secondary) one.
When starting the RDI installation on the second VM, the installer will detect that the RDI
database is already in use and ask you to confirm that you intend to install RDI with HA.

After the installation is complete, you must set the source and target database secrets
on both VMs as described in [Deploy a pipeline](). If you use `redis-di` to deploy your configuration, you only need to do this on one of the VMs, not both.

In a High Availability setup, the RDI pipeline is only active on the primary instance (VM).
The two RDI instances will use the RDI database for leader election. If the primary instance fails
to renew the lease in the RDI database, it will lose the leadership and a failover to the secondary instance
will take place. After the failover, the secondary instance will become the primary one,
and the RDI pipeline will be active on that VM.

You may find it useful to trigger a failover deliberately to check that RDI is correctly configured to handle it. See [Test HA failover]() to learn how to do this.

Prepare your source database

Deploy a pipeline

When the installation is complete, and you have prepared the source database for CDC,
you are ready to start using RDI. See the guides on how to
[configure]() and
[deploy]()
RDI pipelines for more information. You can also configure and deploy a pipeline
using [Redis Insight](). See
[RDI in Redis Insight]()
for full details on how to connect to RDI and deploy pipelines.

Uninstall RDI

If you want to remove your RDI installation, go to the installation folder and run
the uninstall script as a privileged user:

[code example]

The script will ask if you are sure before proceeding:

[code example]

If you type anything other than "y" here, the script will abort without making any changes
to RDI or your source database.

sudo iptables-save | awk '/^[*]/ { print $1 }
                     /^:[A-Z]+ [^-]/ { print $1 " ACCEPT" ; }
                     /COMMIT/ { print $0; }' | sudo iptables-restore

sudo update-alternatives --set iptables /usr/sbin/iptables-legacy
sudo update-alternatives --set ip6tables /usr/sbin/ip6tables-legacy

sudo firewall-cmd --permanent --add-port=443/tcp # RDI API
sudo firewall-cmd --permanent --add-port=6443/tcp # kube-apiserver
sudo firewall-cmd --permanent --zone=trusted --add-source=10.42.0.0/16 # Kubernetes pods
sudo firewall-cmd --permanent --zone=trusted --add-source=10.43.0.0/16 # Kubernetes services
sudo firewall-cmd --reload

sudo systemctl disable nm-cloud-setup.service nm-cloud-setup.timer
sudo reboot

sudo ufw allow 443/tcp # RDI API
sudo ufw allow 6443/tcp # kube-apiserver
sudo ufw allow from 10.42.0.0/16 to any # Kubernetes pods
sudo ufw allow from 10.43.0.0/16 to any # Kubernetes services
sudo ufw reload

export RDI_VERSION=
    wget https://redis-enterprise-software-downloads.s3.amazonaws.com/redis-di/rdi-installation-$RDI_VERSION.tar.gz
    tar -xvf rdi-installation-$RDI_VERSION.tar.gz

sudo ./install.sh --installation-dir <custom-installation-directory>

sudo k3s kubectl delete nodes --all
sudo service k3s restart

This will uninstall RDI and its dependencies, are you sure? [y, N]


Requirements summary

Overview

The sections below summarize the software and hardware requirements for
an RDI installation.

Hardware requirements for VM installation

OS requirements for VM installation

Kubernetes/OpenShift supported versions

RDI database requirements

Supported source databases


Upgrading RDI

Upgrading a VM installation

Follow the steps below to upgrade an existing
[VM installation]()
of RDI:

1.  Download the RDI installer from the [Redis download center](https://redis-enterprise-software-downloads.s3.amazonaws.com/redis-di/rdi-installation-.tar.gz)
    (in the *Modules, Tools & Integration* category) and extract it to your
    preferred installation folder.

    [code example]

1.  Go to the installation folder:

    [code example]

1.  Run the `upgrade.sh` script as a privileged user. Note that you must pass
    your RDI password to the script unless the password is empty.

    [code example]

Recovering from failure during a VM upgrade

If the previous version is v1.4.4 or later, go to the `rdi_install/<PREVIOUS_VERSION>`
directory and run `sudo ./upgrade.sh` to revert to that version, as described in the section
[Upgrading a VM installation](#upgrading-a-vm-installation) above.

If the version you are replacing is earlier than v1.4.4, follow these steps:

1.  Run `redis-di --version` to check the current version.

    If the version is the new one, copy the previous version
    of the RDI CLI to `/usr/local/bin` with the following command:

    [code example]

1.  Check that the CLI version is correct by running `redis-di --version`.

    Then, go to the `rdi_install/<PREVIOUS_VERSION>` directory and run the
    following command;

    [code example]

If the `collector-source` or the `processor` pods are not in the `Running` state after
the upgrade, you must run `redis-di deploy` and check again that they are both in the
`Running` state.

Upgrading a VM installation with High Availability

If there is an active pipeline, upgrade RDI on the active VM first.
This will cause a short pipeline downtime of up to two minutes.
Afterwards, upgrade RDI on the passive VM. This will not cause any downtime.

When upgrading from RDI < 1.8.0 to RDI >= 1.8.0 in a VM HA setup, both RDI instances may incorrectly consider themselves active after the upgrade. This occurs because the upgrade process doesn't change the cluster id value from its default `cluster-1`, causing both clusters to assume they are the active cluster.

**Symptoms:**

- The upgraded passive node will start collector and processor components
- Collector may enter a crash loop as it fails to connect to the source
- Both clusters will restart in a loop

**Workaround:**

After upgrading, manually set a unique cluster ID for one of the installations (preferably on the passive instance):

1. Locate the RDI configuration file on the VM host. The file is typically located at `/etc/rdi/rdi-sys-config.yaml`.
2. Open the configuration file in a text editor. For example:

   [code example]bash
    export RDI_VERSION=
    wget https://redis-enterprise-software-downloads.s3.amazonaws.com/redis-di/rdi-$RDI_VERSION.tgz
    [code example]bash
    helm upgrade --install rdi rdi-<tag>.tar.gz -f rdi-values.yaml -n rdi
    [code example]bash
kubectl get all -n rdi
```

If you find that the upgrade did not work as expected for any reason,
then run the `helm upgrade` command again (as described in the section
[Upgrading a Kubernetes installation](#upgrading-a-kubernetes-installation) above),
but this time with the previous version you were upgrading from, and using
your saved `rdi-values.yaml` for that version. This will restore your previous working state.

Downgrading from RDI 1.8.0 or later to an earlier version using `helm upgrade`
will not work. If you need to perform such an upgrade, uninstall RDI completely first as
described in [Uninstall RDI](),
and then install the old version.

What happens during the upgrade?

The upgrade process replaces the current RDI components with their new versions:

-   Firstly, the control plane components are replaced. At this point, the pipeline
    is still active but monitoring will be disconnected.
-   Secondly, the pipeline data plane components are replaced.
    If a pipeline is active while upgrading, the `collector-source` and `processor`
    pods will be restarted. The pipeline will pause for up to two minutes but it
    will catch up very quickly after restarting.
    The pipeline data and state are both stored in Redis, so data will not
    be lost during the upgrade.

sudo ./upgrade.sh --rdi-password <redis-rdi-password>

sudo cp rdi_install/<PREVIOUS_VERSION>/deps/rdi-cli/<OS>/redis-di usr/local/bin

sudo redis-di upgrade --rdi-host <RDI_REDIS_HOST> --rdi-port <RDI_REDIS_PORT>

sudo nano /etc/rdi/rdi-sys-config.yaml

## Upgrading a Kubernetes installation

Follow the steps below to upgrade an existing
[Kubernetes]()
installation of RDI:

1.  If you are using a private registry, pull the new versions of all images listed in
    [Using a private image registry]()
    and add them to your local registry.

1.  Download the RDI Helm chart tar file from the [Redis download center](https://redis-enterprise-software-downloads.s3.amazonaws.com/redis-di/rdi-.tgz)
    (in the *Modules, Tools & Integration* category).

1.  Adapt your `rdi-values.yaml` file to any changes in the new RDI version if needed.
    See also [Upgrading to RDI 1.8.0 or later from an earlier version](#upgrading-to-rdi-180-or-later-from-an-earlier-version).
    Before making any changes, save your existing `rdi-values.yaml` if you need to revert
    to the old RDI version for any reason.

1.  Run the `helm upgrade` command:

Note that you don't need to
[deploy]()
the RDI configuration again after this step.

### Upgrading to RDI 1.8.0 or later from an earlier version

When upgrading to RDI 1.8.0 or later from an earlier version
you must adapt your `rdi-values.yaml` file to the following changes:

-   All collector and processor values that were previously under `collector`,
    `collectorSourceMetricsExporter`, and `processor` have been moved to
    `operator.dataPlane.collector` and `operator.dataPlane.processor`.
-   `global.collectorApiEnabled` has been moved to `operator.dataPlane.collectorApi.enabled`,
    and is now a boolean value, not `"0"` or `"1"`.
-   `api.authEnabled` is also now a boolean value, not `"0"` or `"1"`.
-   The following values have been deprecated: `rdiMetricsExporter.service.protocol`,
    `rdiMetricsExporter.service.port`, `rdiMetricsExporter.serviceMonitor.path`,
    `api.service.name`.

### Verifying the upgrade

Check that all pods have `Running` status: