Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ weight: 35
---

RDI ships with two stream processor implementations. The *classic*
processor is implemented in Python. The *Flink* processor is built on top of
processor is implemented in Python. The default *Flink* processor is built on top of
[Apache Flink](https://flink.apache.org/). Both run on VM and Kubernetes
installations. The Flink processor can achieve much higher throughput
during snapshots, scales horizontally by changing the number of TaskManager replicas,
Expand Down Expand Up @@ -99,11 +99,9 @@ redis-di deploy --dir <pipeline-config-directory>

Wait for the deployment to finish and the source collector to stop. Keep
the pipeline active so the classic processor can process the remaining input records.
Do not use `redis-di stop` for this step, because it also stops the processor.

Applications can continue writing to the source database while collection
is disabled. Make sure the database change log retains the whole paused
interval. When the collector restarts, it resumes from the saved source
is disabled. When the collector restarts, it resumes from the saved source
position and processes those changes.

## Step 3: Wait for the input streams to empty
Expand All @@ -118,14 +116,11 @@ Do not include DLQ streams. Use any of the following methods.
Run:

```bash
redis-di describe default
redis-di describe
```

In the **Statistics** table, the **Pending** value for each classic processor
stream is its current length. Confirm that every input stream is listed and
that the values agree with the Redis command checks below. This **Pending**
value is different from consumer-group pending entries. `XPENDING` or group
lag of `0` alone does not prove that a stream is empty.
stream is its current length. Confirm that every input stream is listed.

### Check with Redis commands

Expand Down Expand Up @@ -156,24 +151,6 @@ XLEN <input-stream-key>

Run a complete `SCAN` and all `XLEN` commands in each of the three checks.

### Check with Redis Insight

1. Connect Redis Insight to the RDI database and open **Browse**.
1. Filter by the pipeline's input stream pattern. For the default pipeline,
use `data:{rdi}:*`. Confirm that all input streams are listed.
1. Open each stream, select **Stream Data**, and use the refresh button.
Confirm that **Entries** is `0`.
1. Repeat the complete inventory and entry check three times, five seconds
apart.

You can also open the built-in **CLI** and run the `SCAN` and `XLEN` commands
shown above. The Browser and CLI results must contain the same streams and
lengths.

If records remain, keep the classic processor running and resolve its
processing errors before continuing. Do not delete stream entries, reset the
pipeline, or move consumer-group positions to make the count reach `0`.

## Step 4: Switch processors and resume collection

After the drain check passes, remove the source's `active: false` setting
Expand Down Expand Up @@ -262,7 +239,7 @@ for the `ServiceMonitor` configuration and the available metrics.

## Rolling back

To revert a pipeline to the classic processor, set `processors.type` back to
To revert a pipeline to the classic processor, set `processors.type` to
`classic` and redeploy the pipeline. This setting is required on RDI 2.0.0,
where the default is `flink`. The classic processor silently ignores
`processors.advanced`, so you don't need to remove it before switching back.
26 changes: 5 additions & 21 deletions content/integrate/redis-data-integration/installation/upgrade.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,23 +21,6 @@ Before upgrading to RDI 2.0.0, review the
[processor default change](#upgrading-to-rdi-200).
{{< /note >}}

## Upgrading to RDI 2.0.0

RDI 2.0.0 changes the default processor from `classic` to `flink`. This default
applies when the pipeline's `config.yaml` omits `processors.type`.

For an existing classic pipeline, choose one of these options before upgrading:

- To migrate to Flink, first follow
[Migrate from the classic processor to the Flink processor]({{< relref "/integrate/redis-data-integration/installation/migration-classic-to-flink" >}})
on RDI 1.19.0. This stops the collector and drains the input streams before
switching processors. Then upgrade RDI.
- To keep the classic processor, set `processors.type: classic` in the
pipeline's `config.yaml` and deploy it before upgrading.

If your pipeline already uses `processors.type: flink`, no processor change
is needed. Continue with the upgrade instructions for your installation.

## Upgrading a VM installation

Follow the steps below to upgrade an existing
Expand Down Expand Up @@ -254,11 +237,12 @@ The
fully supported on both VM and Kubernetes installations after upgrading to
RDI 1.19.0. Once the upgrade completes, it is always available —
no opt-in is required, and the defaults are sized for typical workloads.
On RDI 1.19.0, existing classic pipelines keep using that processor until
you switch them by setting

{{< warning >}}The Flink processor is the default as of RDI 2.0.0.
Upgrading to that release or later moves a pipeline whose `config.yaml` does not set
[`processors.type`]({{< relref "/integrate/redis-data-integration/data-pipelines/pipeline-config#processors" >}})
to `flink` in their `config.yaml`.
RDI 2.0.0 changes this default; see [Upgrading to RDI 2.0.0](#upgrading-to-rdi-200).
onto the Flink processor when you next deploy it. To keep such a pipeline on the classic
processor, set `processors.type` to `classic` before you upgrade.{{< /warning >}}

On Kubernetes, to override the Flink processor defaults, add an
`operator.dataPlane.flinkProcessor` block to your `rdi-values.yaml` file as
Expand Down
Loading