-
Notifications
You must be signed in to change notification settings - Fork 366
RDSC-6045: Document draining streams before Flink processor migration #4017
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
492e76d
cbdb948
72c6155
2346af8
7181547
37ed611
35d1366
57c2c73
a0c45aa
be75689
d1b43b9
059c44c
93e57c8
bb55b9e
3da5154
c9e8a95
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -16,26 +16,42 @@ type: integration | |||||||||||||||||||||||||||||||||||
| weight: 35 | ||||||||||||||||||||||||||||||||||||
| --- | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| RDI ships with two stream processor implementations. The default *classic* | ||||||||||||||||||||||||||||||||||||
| RDI ships with two stream processor implementations. The *classic* | ||||||||||||||||||||||||||||||||||||
| processor is implemented in Python. The *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, | ||||||||||||||||||||||||||||||||||||
| and uses Flink checkpointing for fault tolerance. See [Stream processor implementations]({{< relref "/integrate/redis-data-integration/architecture#stream-processor-implementations" >}}) | ||||||||||||||||||||||||||||||||||||
| for an overview. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| The classic processor is the default in RDI 1.19.0. The Flink processor is the | ||||||||||||||||||||||||||||||||||||
| default starting with RDI 2.0.0. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| This page describes how to migrate an existing pipeline from the classic | ||||||||||||||||||||||||||||||||||||
| processor to the Flink processor. The steps are the same on VMs and Kubernetes, | ||||||||||||||||||||||||||||||||||||
| except for the optional Helm-level tuning in [Step 1](#step-1-configure-the-flink-processor-at-the-helm-chart-level-kubernetes), | ||||||||||||||||||||||||||||||||||||
| which applies to Kubernetes only. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ## Before you migrate | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| This procedure migrates the pipeline processor on RDI 1.19.0. It does not | ||||||||||||||||||||||||||||||||||||
| upgrade RDI. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Before you start, save your existing configuration and jobs. Wait for the | ||||||||||||||||||||||||||||||||||||
| initial snapshot to finish. Interrupting it causes the snapshot to restart | ||||||||||||||||||||||||||||||||||||
| from the beginning. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| {{< warning >}} | ||||||||||||||||||||||||||||||||||||
| Switching processors with records still in the RDI input streams can leave | ||||||||||||||||||||||||||||||||||||
| records unprocessed. Stop the collector and let the classic processor empty | ||||||||||||||||||||||||||||||||||||
| the streams before switching. | ||||||||||||||||||||||||||||||||||||
| {{< /warning >}} | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Confirm that your pipeline is compatible with the Flink processor: | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| - `JSON.MERGE` semantics differ from the classic processor's Lua-based merge | ||||||||||||||||||||||||||||||||||||
| when null values are involved (see | ||||||||||||||||||||||||||||||||||||
| [`use_native_json_merge`]({{< relref "/integrate/redis-data-integration/reference/config-yaml-reference#processors" >}})). | ||||||||||||||||||||||||||||||||||||
| [`use_native_json_merge`]({{< relref "/integrate/redis-data-integration/reference/config-yaml-reference#processors-data-processing-configuration" >}})). | ||||||||||||||||||||||||||||||||||||
| The Flink processor always uses the native `JSON.MERGE` command when the | ||||||||||||||||||||||||||||||||||||
| target database supports it. | ||||||||||||||||||||||||||||||||||||
| - Ensure your Kubernetes cluster or VM has enough capacity for the Flink JobManager | ||||||||||||||||||||||||||||||||||||
|
|
@@ -46,7 +62,7 @@ Confirm that your pipeline is compatible with the Flink processor: | |||||||||||||||||||||||||||||||||||
| ## Step 1: Configure the Flink processor at the Helm chart level (Kubernetes) | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| This step applies to **Kubernetes** installations only. On VM installations, | ||||||||||||||||||||||||||||||||||||
| skip it and enable the Flink processor per pipeline in step 2. | ||||||||||||||||||||||||||||||||||||
| continue with [Step 2](#step-2-disable-source-collection). | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| The Flink processor is always available — no opt-in is required at the Helm | ||||||||||||||||||||||||||||||||||||
| chart level. The defaults are sized for typical workloads, so you can skip | ||||||||||||||||||||||||||||||||||||
|
|
@@ -55,27 +71,137 @@ TaskManager defaults, add an `operator.dataPlane.flinkProcessor` block to | |||||||||||||||||||||||||||||||||||
| your `rdi-values.yaml` file and run `helm upgrade` as described in | ||||||||||||||||||||||||||||||||||||
| [Configure the Flink processor]({{< relref "/integrate/redis-data-integration/installation/install-k8s#configure-the-flink-processor" >}}). | ||||||||||||||||||||||||||||||||||||
| Existing pipelines continue to run on the classic processor until you switch | ||||||||||||||||||||||||||||||||||||
| them in step 2. | ||||||||||||||||||||||||||||||||||||
| them in [Step 4](#step-4-switch-processors-and-resume-collection). | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| For VM installations, skip this step. You can configure per-pipeline Flink | ||||||||||||||||||||||||||||||||||||
| resources in step 4. | ||||||||||||||||||||||||||||||||||||
| resources in [Step 6](#step-6-tune-the-flink-processor-optional). | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ## Step 2: Disable source collection | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| In your existing `config.yaml`, add `active: false` under the source and set | ||||||||||||||||||||||||||||||||||||
| `processors.type` to `classic`. Use your existing source name and preserve | ||||||||||||||||||||||||||||||||||||
| all other source, target, processor, and job settings. This example shows | ||||||||||||||||||||||||||||||||||||
| only the fields to change: | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ```yaml | ||||||||||||||||||||||||||||||||||||
| sources: | ||||||||||||||||||||||||||||||||||||
| <existing-source-name>: | ||||||||||||||||||||||||||||||||||||
| active: false | ||||||||||||||||||||||||||||||||||||
| processors: | ||||||||||||||||||||||||||||||||||||
| type: classic | ||||||||||||||||||||||||||||||||||||
| ``` | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Deploy the complete configuration directory, including the existing jobs: | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ```bash | ||||||||||||||||||||||||||||||||||||
| 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. | ||||||||||||||||||||||||||||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| 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 | ||||||||||||||||||||||||||||||||||||
| position and processes those changes. | ||||||||||||||||||||||||||||||||||||
|
Comment on lines
+104
to
+107
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ## Step 3: Wait for the input streams to empty | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| After the collector has stopped, wait for every input stream to have a length | ||||||||||||||||||||||||||||||||||||
| of `0` in three complete checks, five seconds apart. An error, missing | ||||||||||||||||||||||||||||||||||||
| statistics, or an unexpectedly empty stream list does not count as `0`. | ||||||||||||||||||||||||||||||||||||
| Do not include DLQ streams. Use any of the following methods. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ### Check with `redis-di` | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Run: | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ```bash | ||||||||||||||||||||||||||||||||||||
| redis-di describe default | ||||||||||||||||||||||||||||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||||||||||||||||||||||||||||||||
| ``` | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ## Step 2: Switch the pipeline to the Flink processor | ||||||||||||||||||||||||||||||||||||
| 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. | ||||||||||||||||||||||||||||||||||||
|
Comment on lines
+124
to
+128
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| In the pipeline's `config.yaml`, set | ||||||||||||||||||||||||||||||||||||
| ### Check with Redis commands | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Connect an authenticated Redis client to the **RDI database** that stores | ||||||||||||||||||||||||||||||||||||
| the pipeline's input streams, not the target database. Find the input stream | ||||||||||||||||||||||||||||||||||||
| keys with | ||||||||||||||||||||||||||||||||||||
| [`SCAN`]({{< relref "/commands/scan" >}}): | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ```text | ||||||||||||||||||||||||||||||||||||
| SCAN 0 MATCH data:{rdi}:* COUNT 1000 TYPE stream | ||||||||||||||||||||||||||||||||||||
| ``` | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| If the returned cursor is not `0`, pass it to the next command: | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ```text | ||||||||||||||||||||||||||||||||||||
| SCAN <returned-cursor> MATCH data:{rdi}:* COUNT 1000 TYPE stream | ||||||||||||||||||||||||||||||||||||
| ``` | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Repeat with each new cursor until the returned cursor is `0`, even if an | ||||||||||||||||||||||||||||||||||||
| intermediate result contains no keys. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| For every input stream returned, run | ||||||||||||||||||||||||||||||||||||
| [`XLEN`]({{< relref "/commands/xlen" >}}): | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ```text | ||||||||||||||||||||||||||||||||||||
| 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`. | ||||||||||||||||||||||||||||||||||||
|
Comment on lines
+159
to
+175
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ## Step 4: Switch processors and resume collection | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| After the drain check passes, remove the source's `active: false` setting | ||||||||||||||||||||||||||||||||||||
| from the existing `config.yaml` and set | ||||||||||||||||||||||||||||||||||||
| [`processors.type`]({{< relref "/integrate/redis-data-integration/data-pipelines/pipeline-config#processors" >}}) | ||||||||||||||||||||||||||||||||||||
| to `flink`: | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ```yaml | ||||||||||||||||||||||||||||||||||||
| processors: | ||||||||||||||||||||||||||||||||||||
| type: flink | ||||||||||||||||||||||||||||||||||||
| ... | ||||||||||||||||||||||||||||||||||||
| ``` | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Then redeploy the pipeline. The operator stops the classic processor pods | ||||||||||||||||||||||||||||||||||||
| and starts the Flink JobManager and TaskManager workloads for the pipeline. | ||||||||||||||||||||||||||||||||||||
| RDI 1.19.0 requires this setting because its default processor is `classic`. | ||||||||||||||||||||||||||||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Keep the remaining configuration and jobs, then redeploy the complete | ||||||||||||||||||||||||||||||||||||
| configuration directory: | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ```bash | ||||||||||||||||||||||||||||||||||||
| redis-di deploy --dir <pipeline-config-directory> | ||||||||||||||||||||||||||||||||||||
| ``` | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Wait for the classic processor to terminate and the Flink JobManager and | ||||||||||||||||||||||||||||||||||||
| TaskManager workloads to become healthy. Confirm that collection resumes | ||||||||||||||||||||||||||||||||||||
| from the saved source position and changes committed during the pause reach | ||||||||||||||||||||||||||||||||||||
| the target. Verify new inserts, updates, and deletes. The processor migration | ||||||||||||||||||||||||||||||||||||
| is complete after these checks pass. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ## Step 3: Adapt deprecated and classic-only properties | ||||||||||||||||||||||||||||||||||||
| ## Step 5: Adapt deprecated and classic-only properties | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Some `processors` properties are no-ops, classic-only, or have moved to | ||||||||||||||||||||||||||||||||||||
| `processors.advanced` for the Flink processor. The following table lists the | ||||||||||||||||||||||||||||||||||||
|
|
@@ -97,7 +223,7 @@ and the Flink processor silently ignores classic-only top-level properties, so k | |||||||||||||||||||||||||||||||||||
| both top-level properties and their `processors.advanced` equivalents lets | ||||||||||||||||||||||||||||||||||||
| you switch back without further edits. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ## Step 4: Tune the Flink processor (optional) | ||||||||||||||||||||||||||||||||||||
| ## Step 6: Tune the Flink processor (optional) | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| Fine-tune the Flink processor through the `processors.advanced` section. | ||||||||||||||||||||||||||||||||||||
| For example: | ||||||||||||||||||||||||||||||||||||
|
|
@@ -121,10 +247,10 @@ processors: | |||||||||||||||||||||||||||||||||||
| ``` | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| See the | ||||||||||||||||||||||||||||||||||||
| [`processors.advanced` reference]({{< relref "/integrate/redis-data-integration/reference/config-yaml-reference#processors" >}}) | ||||||||||||||||||||||||||||||||||||
| [`processors.advanced` reference]({{< relref "/integrate/redis-data-integration/reference/config-yaml-reference#processorsadvanced-advanced-configuration" >}}) | ||||||||||||||||||||||||||||||||||||
| for the full set of available properties. | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| ## Step 5: Update observability | ||||||||||||||||||||||||||||||||||||
| ## Step 7: Update observability | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| The Flink processor exposes Prometheus metrics directly | ||||||||||||||||||||||||||||||||||||
| from the Flink JobManager and TaskManager pods. | ||||||||||||||||||||||||||||||||||||
|
|
@@ -135,6 +261,6 @@ for the `ServiceMonitor` configuration and the available metrics. | |||||||||||||||||||||||||||||||||||
| ## Rolling back | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| To revert a pipeline to the classic processor, set `processors.type` back to | ||||||||||||||||||||||||||||||||||||
| `classic` (or remove the property) and redeploy the pipeline. The | ||||||||||||||||||||||||||||||||||||
| `processors.advanced` section is silently ignored by the classic processor, | ||||||||||||||||||||||||||||||||||||
| so you don't need to remove it before switching back. | ||||||||||||||||||||||||||||||||||||
| `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. | ||||||||||||||||||||||||||||||||||||
Uh oh!
There was an error while loading. Please reload this page.