Repository: flink-web Updated Branches: refs/heads/asf-site cde075c50 -> 035337bd3
Add Incremental Checkpointing blog post Project: http://git-wip-us.apache.org/repos/asf/flink-web/repo Commit: http://git-wip-us.apache.org/repos/asf/flink-web/commit/8f98e453 Tree: http://git-wip-us.apache.org/repos/asf/flink-web/tree/8f98e453 Diff: http://git-wip-us.apache.org/repos/asf/flink-web/diff/8f98e453 Branch: refs/heads/asf-site Commit: 8f98e4537b2b73d9f42d489412e8ceac1d83d173 Parents: cde075c Author: wints <[email protected]> Authored: Tue Jan 30 18:38:37 2018 +0100 Committer: Fabian Hueske <[email protected]> Committed: Wed Feb 7 18:57:08 2018 +0100 ---------------------------------------------------------------------- _posts/2018-01-30-incremental-checkpointing.md | 102 ++++++++++++++++++++ img/blog/incremental_cp_impl_example.svg | 3 + 2 files changed, 105 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink-web/blob/8f98e453/_posts/2018-01-30-incremental-checkpointing.md ---------------------------------------------------------------------- diff --git a/_posts/2018-01-30-incremental-checkpointing.md b/_posts/2018-01-30-incremental-checkpointing.md new file mode 100644 index 0000000..d9ee2ac --- /dev/null +++ b/_posts/2018-01-30-incremental-checkpointing.md @@ -0,0 +1,102 @@ +--- +layout: post +title: 'Managing Large State in Apache Flink: An Intro to Incremental Checkpointing' +date: 2018-01-30T12:00:00.000Z +authors: +- stefan: + name: "Stefan Ricther" + twitter: "StefanRRicther" +- chris: + name: "Chris Ward" + twitter: "chrischinch" +excerpt: Flink 1.3.0 introduced incremental checkpointing, making it possible for applications with large state to generate checkpoints more efficiently. +categories: features +--- + +Apache Flink was purpose-built for _stateful_ stream processing. However, what is state in a stream processing application? I defined state and stateful stream processing in a [previous blog post](http://flink.apache.org/features/2017/07/04/flink-rescalable-state.html), and in case you need a refresher, _state is defined as memory in an application's operators that stores information about previously-seen events that you can use to influence the processing of future events_. + +State is a fundamental, enabling concept in stream processing required for a majority of complex use cases. Some examples highlighted in the [Flink documentation](https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/stream/state.html): + +- When an application searches for certain event patterns, the state stores the sequence of events encountered so far. +- When aggregating events per minute, the state holds the pending aggregates. +- When training a machine learning model over a stream of data points, the state holds the current version of the model parameters. + +However, stateful stream processing is only useful in production environments if the state is fault tolerant. "Fault tolerance" means that even if there's a software or machine failure, the computed end-result is accurate, with no data loss or double-counting of events. + +Flink's fault tolerance has always been a powerful and popular feature, minimizing the impact of software or machine failure on your business and making it possible to guarantee exactly-once results from a Flink application. + +Core to this is [checkpointing](https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/stream/checkpointing.html), which is the mechanism Flink uses to make application state fault tolerant. A checkpoint in Flink is a global, asynchronous snapshot of application state that's taken on a regular interval and sent to durable storage (usually, a distributed file system). In the event of a failure, Flink restarts an application using the most recently completed checkpoint as a starting point. Some Apache Flink users run applications with gigabytes or even terabytes of application state. These users reported that with such large state, creating a checkpoint was often a slow and resource intensive operation, which is why in Flink 1.3 we introduced 'incremental checkpointing.' + +Before incremental checkpointing, every single Flink checkpoint consisted of the full state of an application. We created the incremental checkpointing feature after we noticed that writing the full state for every checkpoint was often unnecessary, as the state changes from one checkpoint to the next were rarely that large. Incremental checkpointing instead maintains the differences (or 'delta') between each checkpoint and stores only the differences between the last checkpoint and the current state. + +Incremental checkpoints can provide a significant performance improvement for jobs with a very large state. Early testing of the feature by a production user with terabytes of state shows a drop in checkpoint time from more than 3 minutes down to 30 seconds after implementing incremental checkpoints. This is because the checkpoint doesn't need to transfer the full state to durable storage on each checkpoint. + +### How to Start + +Currently, you can only use incremental checkpointing with a RocksDB state back-end, and Flink uses RocksDB's internal backup mechanism to consolidate checkpoint data over time. As a result, the incremental checkpoint history in Flink does not grow indefinitely, and Flink eventually consumes and prunes old checkpoints automatically. + +To enable incremental checkpointing in your application, I recommend you read the [the Apache Flink documentation on checkpointing](https://ci.apache.org/projects/flink/flink-docs-release-1.4/ops/state/large_state_tuning.html#tuning-rocksdb) for full details, but in summary, you enable checkpointing as normal, but enable incremental checkpointing in the constructor by setting the second parameter to `true`. + +#### Java Example + +```java +StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); +env.setStateBackend(new RocksDBStateBackend(filebackend, true)); +``` + +#### Scala Example + +```scala +val env = StreamExecutionEnvironment.getExecutionEnvironment() +env.setStateBackend(new RocksDBStateBackend(filebackend, true)) +``` + +By default, Flink retains 1 completed checkpoint, so if you need a higher number, [you can configure it with the following flag](https://ci.apache.org/projects/flink/flink-docs-master/dev/stream/state/checkpointing.html#related-config-options): + +```java +state.checkpoints.num-retained +``` + +### How it Works + +Flink's incremental checkpointing uses [RocksDB checkpoints](https://github.com/facebook/rocksdb/wiki/Checkpoints) as a foundation. RocksDB is a key-value store based on '[log-structured-merge](https://en.wikipedia.org/wiki/Log-structured_merge-tree)' (LSM) trees that collects all changes in a mutable (changeable) in-memory buffer called a 'memtable'. Any updates to the same key in the memtable replace previous values, and once the memtable is full, RocksDB writes it to disk with all entries sorted by their key and with light compression applied. Once RocksDB writes the memtable to disk it is immutable (unchangeable) and is now called a 'sorted-string-table' (sstable). + +A 'compaction' background task merges sstables to consolidate potential duplicates for each key, and over time RocksDB deletes the original sstables, with the merged sstable containing all information from across all the other sstables. + +On top of this, Flink tracks which sstable files RocksDB has created and deleted since the previous checkpoint, and as the sstables are immutable, Flink uses this to figure out the state changes. To do this, Flink triggers a flush in RocksDB, forcing all memtables into sstables on disk, and hard-linked in a local temporary directory. This process is synchronous to the processing pipeline, and Flink performs all further steps asynchronously and does not block processing. + +Then Flink copies all new sstables to stable storage (e.g., HDFS, S3) to reference in the new checkpoint. Flink doesn't copy all sstables that already existed in the previous checkpoint to stable storage but re-reference them. Any new checkpoints will no longer reference deleted files as deleted sstables in RocksDB are always the result of compaction, and it eventually replaces old tables with an sstable that is the result of a merge. This how in Flink's incremental checkpoints can prune the checkpoint history. + +For tracking changes between checkpoints, the uploading of consolidated tables is redundant work. Flink performs the process incrementally, and typically adds only a small overhead, so we consider this worthwhile because it allows Flink to keep a shorter history of checkpoints to consider in a recovery. + +#### An Example + + +_Example setup_ + +Take an example with a subtask of one operator that has a keyed state, and the number of retained checkpoints set at **2**. The columns in the figure above show the state of the local RocksDB instance for each checkpoint, the files it references, and the counts in the shared state registry after the checkpoint completes. + +For checkpoint 'CP 1', the local RocksDB directory contains two sstable files, it considers these new and uploads them to stable storage using directory names that match the checkpoint name. When the checkpoint completes, Flink creates the two entries in the shared state registry and sets their counts to '1'. The key in the shared state registry is a composite of an operator, subtask, and the original sstable file name. The registry also keeps a mapping from the key to the file path in stable storage. + +For checkpoint 'CP 2', RocksDB has created two new sstable files, and the two older ones still exist. For checkpoint 'CP 2', Flink adds the two new files to stable storage and can reference the previous two files. When the checkpoint completes, Flink increases the counts for all referenced files by 1. + +For checkpoint 'CP 3', RocksDB's compaction has merged `sstable-(1)`, `sstable-(2)`, and `sstable-(3)` into `sstable-(1,2,3)` and deleted the original files. This merged file contains the same information as the source files, with all duplicate entries eliminated. In addition to this merged file, `sstable-(4)` still exists and there is now a new `sstable-(5)` file. Flink adds the new `sstable-(1,2,3)` and `sstable-(5)` files to stable storage, `sstable-(4)` is re-referenced from checkpoint 'CP 2' and increases the counts for referenced files by 1. The older 'CP 1' checkpoint is now deleted as the number of retained checkpoints (2) has been reached. As part of this deletion, Flink decreases the counts for all files referenced 'CP 1', (`sstable-(1)` and `sstable-(2)`), by 1. + +For checkpoint 'CP-4', RocksDB has merged `sstable-(4)`, `sstable-(5)`, and a new `sstable-(6)` into `sstable-(4,5,6)`. Flink adds this new table to stable storage and references it together with `sstable-(1,2,3)`, it increases the counts for `sstable-(1,2,3)` and `sstable-(4,5,6)` by 1 and then deletes 'CP-2' as the number of retained checkpoints has been reached. As the counts for `sstable-(1)`, `sstable-(2)`, and `sstable-(3)` have now dropped to 0, and Flink deletes them from stable storage. + +### Race Conditions and Concurrent Checkpoints + +As Flink can execute multiple checkpoints in parallel, sometimes new checkpoints start before confirming previous checkpoints as completed. Because of this, you should consider which the previous checkpoint to use as a basis for a new incremental checkpoint. Flink only references state from a checkpoint confirmed by the checkpoint coordinator so that it doesn't unintentionally reference a deleted shared file. + +### Restoring Checkpoints and Performance Considerations + +If you enable incremental checkpointing, there are no further configuration steps needed to recover your state in case of failure. If a failure occurs, Flink's `JobManager` tells all tasks to restore from the last completed checkpoint, be it a full or incremental checkpoint. Each `TaskManager` then downloads their share of the state from the checkpoint on the distributed file system. + +Though the feature can lead to a substantial improvement in checkpoint time for users with a large state, there are trade-offs to consider with incremental checkpointing. Overall, the process reduces the checkpointing time during normal operations but can lead to a longer recovery time depending on the size of your state. If the cluster failure is particularly severe and the Flink `TaskManager`s have to read from multiple checkpoints, recovery can be a slower operation than when using non-incremental checkpointing. You can also no longer delete old checkpoints as newer checkpoints need them, and the history of differences between checkpoints can grow indefinitely over time. You need to plan for larger distributed storage to maintain the checkpoints and the network overhead to read from it. + +There are some strategies for improving the convenience/performance trade-off, and I recommend you read [the Flink documentation](https://ci.apache.org/projects/flink/flink-docs-release-1.4/ops/state/checkpoints.html#basics-of-incremental-checkpoints) for more details. + +_This post <a href="https://data-artisans.com/blog/managing-large-state-apache-flink-incremental-checkpointing-overview"; target="_blank"> originally appeared on the data Artisans blog </a>and was contributed to the Flink blog by Stefan Richter and Chris Ward._ +<link rel="canonical" href="https://data-artisans.com/blog/managing-large-state-apache-flink-incremental-checkpointing-overview";> + + http://git-wip-us.apache.org/repos/asf/flink-web/blob/8f98e453/img/blog/incremental_cp_impl_example.svg ---------------------------------------------------------------------- diff --git a/img/blog/incremental_cp_impl_example.svg b/img/blog/incremental_cp_impl_example.svg new file mode 100644 index 0000000..5c852ab --- /dev/null +++ b/img/blog/incremental_cp_impl_example.svg @@ -0,0 +1,3 @@ +<?xml version="1.0"?> +<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd";> +<svg xmlns="http://www.w3.org/2000/svg"; xmlns:xl="http://www.w3.org/1999/xlink"; version="1.1" viewBox="5.456693 3543.4567 789.0866 317.08661" width="789.0866pt" height="317.08661pt"><metadata xmlns:dc="http://purl.org/dc/elements/1.1/";><dc:date>2017-08-08 10:07Z</dc:date><!-- Produced by OmniGraffle Professional 5.4.4 --></metadata><defs><font-face font-family="Monaco" font-size="7" units-per-em="1000" underline-position="-37.597656" underline-thickness="75.683594" slope="0" x-height="560.54688" cap-height="780.27344" ascent="1e3" descent="-250" font-weight="500"><font-face-src><font-face-name name="Monaco"/></font-face-src></font-face><font-face font-family="Helvetica" font-size="12" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="532.22656" cap-height="719.72656" ascent="770.01953" descent="-229.98047" font-weight="bold"><font-face-src><font-face-name name="Helvetica-Bold"/></font-face-src></font-face><font-face font-family=" Helvetica" font-size="9" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="522.94922" cap-height="717.28516" ascent="770.01953" descent="-229.98047" font-weight="500"><font-face-src><font-face-name name="Helvetica"/></font-face-src></font-face><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="DimensionArrow_Marker" viewBox="-1 -5 7 10" markerWidth="7" markerHeight="10" color="#738a05"><g><path d="M 0 0 L 4.8000002 0 M 4.8000002 3.0000001 L 4.8000002 -3.0000001 M 0 1.20000005 L 4.2000002 0 L 0 -1.20000005" fill="none" stroke="currentColor" stroke-width="1"/></g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="DimensionArrow_Marker_2" viewBox="-6 -5 7 10" markerWidth="7" markerHeight="10" color="#738a05"><g><path d="M 0 0 L -4.8000002 0 M -4.8000002 -3.0000001 L -4.8000002 3.0000001 M 0 -1.20000005 L -4.2000002 0 L 0 1.20000005" fill="none" stroke="currentColor" stroke-width="1"/> </g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="DimensionArrow_Marker_3" viewBox="-1 -5 7 10" markerWidth="7" markerHeight="10" color="#bd3612"><g><path d="M 0 0 L 4.8000002 0 M 4.8000002 3.0000001 L 4.8000002 -3.0000001 M 0 1.20000005 L 4.2000002 0 L 0 -1.20000005" fill="none" stroke="currentColor" stroke-width="1"/></g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="DimensionArrow_Marker_4" viewBox="-6 -5 7 10" markerWidth="7" markerHeight="10" color="#bd3612"><g><path d="M 0 0 L -4.8000002 0 M -4.8000002 -3.0000001 L -4.8000002 3.0000001 M 0 -1.20000005 L -4.2000002 0 L 0 1.20000005" fill="none" stroke="currentColor" stroke-width="1"/></g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="DimensionArrow_Marker_5" viewBox="-1 -5 7 10" markerWidth="7" markerHeight="10" color="#2076c8"><g><path d="M 0 0 L 4.8000002 0 M 4.8000002 3.0000001 L 4.8000002 -3.0000001 M 0 1.20000005 L 4.2000002 0 L 0 -1.20000005" fill="none" stroke="currentColor" stroke-width="1"/></g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="DimensionArrow_Marker_6" viewBox="-6 -5 7 10" markerWidth="7" markerHeight="10" color="#2076c8"><g><path d="M 0 0 L -4.8000002 0 M -4.8000002 -3.0000001 L -4.8000002 3.0000001 M 0 -1.20000005 L -4.2000002 0 L 0 1.20000005" fill="none" stroke="currentColor" stroke-width="1"/></g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="DimensionArrow_Marker_7" viewBox="-1 -5 7 10" markerWidth="7" markerHeight="10" color="#a57706"><g><path d="M 0 0 L 4.8000002 0 M 4.8000002 3.0000001 L 4.8000002 -3.0000001 M 0 1.20000005 L 4.2000002 0 L 0 -1.20000005" fill="none" stroke="currentColor" stroke-width="1"/></g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="DimensionArrow_Marker_8" viewBox="-6 -5 7 10" markerWidth="7" markerHeight="10" color="#a57706"><g><path d="M 0 0 L -4.8000002 0 M -4.8000002 -3.0000001 L -4.8000002 3.0000001 M 0 -1.20000005 L -4.2000002 0 L 0 1.20000005" fill="none" stroke="currentColor" stroke-width="1"/></g></marker></defs><g stroke="none" stroke-opacity="1" stroke-dasharray="none" fill="none" fill-opacity="1"><title>Arbeitsfläche 1</title><rect fill="white" width="1118.55115" height="3914.4489"/><g><title>Ebene 1</title><line x1="232.43229" y1="3744.6237" x2="165.44272" y2="3815.9998" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="160.75" y1="3744.624" x2="160.75" y2="3815.9998" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="84.375015" y1="3663.7481" x2="84.375015" y2="3734.6237" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="156.02824" y1="3663.7482" x2="89.09676" y2="3734.6237" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="227.68149" y1="3663.7481" x2="93.81851" y2="3734.6237" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><rect x="49" y="3729.8736" width="70.75" height="18.500217" fill="#0a2832"/><rect x="49" y="3729.8736" width="70.75" height="18.500217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(49 3734.8315)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="3.869873" y="7" textLength="63.010254">sstable-(1,2,3)</tspan></text><rect x="49" y="3589.749" width="70.75" height="18.500217" fill="#0a2832"/><rect x="49" y="3589.749" width="70.75" height="18.500217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(49 3594.707)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="12.27124" y="7" textLength="46.20752">sstable-(1)</tspan></text><rect x="125.875" y="3589.749" width="70.75" height="18.500217" fill="#0a2832" /><rect x="125.875" y="3589.749" width="70.75" height="18.500217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(125.875 3594.707)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="12.27124" y="7" textLength="46.20752">sstable-(2)</tspan></text><rect x="49" y="3648.998" width="70.75" height="18.500217" fill="#0a2832"/><rect x="49" y="3648.998" width="70.75" height="18.500217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(49 3653.9559)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="12.27124" y="7" textLength="46.20752">sstable-(1)</tspan></text><rect x="125.375" y="3648.998" width="70.75" height="18.500217" fill="#0a2832"/><rect x="125.375" y="3648.998" width="70.75" height="18.500217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="tr anslate(125.375 3653.956)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="12.27124" y="7" textLength="46.20752">sstable-(2)</tspan></text><rect x="201.75" y="3648.998" width="70.75" height="18.500217" fill="#0a2832"/><rect x="201.75" y="3648.998" width="70.75" height="18.500217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(201.75 3653.9559)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="12.27124" y="7" textLength="46.20752">sstable-(3)</tspan></text><rect x="278.125" y="3648.998" width="70.75" height="18.500217" fill="#0a2832"/><rect x="278.125" y="3648.998" width="70.75" height="18.500217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(278.125 3653.956)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="12.27124" y="7" textLength="46.20752" >sstable-(4)</tspan></text><rect x="125.375" y="3729.8737" width="70.75" >height="18.500217" fill="#0a2832"/><rect x="125.375" y="3729.8737" >width="70.75" height="18.500217" stroke="black" stroke-linecap="round" >stroke-linejoin="round" stroke-width="1"/><text transform="translate(125.375 >3734.8317)" fill="white"><tspan font-family="Monaco" font-size="7" >font-weight="500" fill="white" x="12.27124" y="7" >textLength="46.20752">sstable-(4)</tspan></text><rect x="201.75" >y="3729.8735" width="70.75" height="18.500217" fill="#0a2832"/><rect >x="201.75" y="3729.8735" width="70.75" height="18.500217" stroke="black" >stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text >transform="translate(201.75 3734.8314)" fill="white"><tspan >font-family="Monaco" font-size="7" font-weight="500" fill="white" >x="12.27124" y="7" textLength="46.20752">sstable-(5)</tspan></text><text >transform="translate(11.5 3591.999)" fill="#738a05"><tspan >font-family="Helvetica" font-size="12" font-weight="bold " fill="#738a05" x=".26660156" y="11" textLength="16.669922">CP</tspan><tspan font-family="Helvetica" font-size="12" font-weight="bold" fill="#738a05" x="16.725586" y="11" textLength="10.0078125"> 1</tspan></text><text transform="translate(11 3651.248)" fill="#bd3612"><tspan font-family="Helvetica" font-size="12" font-weight="bold" fill="#bd3612" x=".26660156" y="11" textLength="16.669922">CP</tspan><tspan font-family="Helvetica" font-size="12" font-weight="bold" fill="#bd3612" x="16.725586" y="11" textLength="10.0078125"> 2</tspan></text><text transform="translate(11 3732.1238)" fill="#2076c8"><tspan font-family="Helvetica" font-size="12" font-weight="bold" fill="#2076c8" x=".26660156" y="11" textLength="16.669922">CP</tspan><tspan font-family="Helvetica" font-size="12" font-weight="bold" fill="#2076c8" x="16.725586" y="11" textLength="10.0078125"> 3</tspan></text><text transform="translate(89.5 3671.686)" fill="black"><tspan font-family="Helvetica" font-size="9" font-weight="500" x=".24487305" y="9" textLength="25.510254">merge</tspan></text><rect x="360.65693" y="3581.999" width="207.84307" height="34.000217" fill="black"/><rect x="360.65693" y="3581.999" width="207.84307" height="34.000217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(365.65693 3587.207)" fill="#738a05"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#738a05" x="0" y="7" textLength="172.22803">op-2-1-sstable-(1) -> dfs://cp-1/23-23-42</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#738a05" x="0" y="21" textLength="29.404785">op-2-1-</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#738a05" x="29.404785" y="21" textLength="63.010254">sstable-(2) -> </tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#738a05" x="92.41504" y="21" textLength="79.81299">dfs://cp-1/54-75-91</tspan></text><rect x="360.65693" y="3625.998" width="207.84308" he ight="65.75202" fill="black"/><rect x="360.65693" y="3625.998" width="207.84308" height="65.75202" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(365.65693 3633.0818)" fill="#738a05"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#738a05" x="0" y="7" textLength="180.62939">op-2-1-sstable-(1) -> (dfs://cp-1/23-23-42)</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#738a05" x="0" y="21" textLength="180.62939">op-2-1-sstable-(2) -> (dfs://cp-1/54-75-91)</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#bd3612" x="0" y="35" textLength="172.22803">op-2-1-sstable-(3) -> dfs://cp-2/44-63-90</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#bd3612" x="0" y="49" textLength="172.22803">op-2-1-sstable-(4) -> dfs://cp-2/41-74-28</tspan></text><rect x="360.65693" y="3702.2488" width="207.84308" height="56.5" fill="black"/><rect x="360 .65693" y="3702.2488" width="207.84308" height="56.5" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(365.65693 3711.7065)" fill="#2076c8"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#2076c8" x="0" y="7" textLength="189.03076">op-2-1-sstable-(1,2,3) -> dfs://cp-3/38-46-83</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#bd3612" x="0" y="21" textLength="180.62939">op-2-1-sstable-(4) -> (dfs://cp-2/41-74-28)</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#2076c8" x="0" y="35" textLength="29.404785">op-2-1-</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#2076c8" x="29.404785" y="35" textLength="155.42529">sstable-(5) -> dfs://cp-3/79-20-41-55</tspan></text><rect x="577.81385" y="3581.999" width="123.62231" height="34.000286" fill="black"/><rect x="577.81385" y="3581.999" width="123.62231" height="34.000286" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(582.81385 3587.2069)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="7" textLength="92.41504">op-2-1-sstable-(1) : 1</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="21" textLength="92.41504">op-2-1-sstable-(2) : 1</tspan></text><rect x="579.28193" y="3625.998" width="122.15423" height="65.75202" fill="black"/><rect x="579.28193" y="3625.998" width="122.15423" height="65.75202" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(584.28193 3633.0818)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="7" textLength="92.41504">op-2-1-sstable-(1) : 2</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="21" textLength="92.41504">op-2-1-sstable-(2) : 2</tspan><tspan font-family ="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="35" textLength="92.41504">op-2-1-sstable-(3) : 1</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="49" textLength="92.41504">op-2-1-sstable-(4) : 1</tspan></text><rect x="579.28193" y="3702.2488" width="122.15423" height="82.25122" fill="black"/><rect x="579.28193" y="3702.2488" width="122.15423" height="82.25122" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(584.28193 3703.5822)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="7" textLength="92.41504">op-2-1-sstable-(1) : 1</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="21" textLength="92.41504">op-2-1-sstable-(2) : 1</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="35" textLength="92.41504">op-2-1-sstable-(3) : 1</tspan><tspan font-family="Mon aco" font-size="7" font-weight="500" fill="white" x="0" y="49" textLength="109.217773">op-2-1-sstable-(1,2,3) : 1</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="63" textLength="92.41504">op-2-1-sstable-(4) : 2</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="77" textLength="92.41504">op-2-1-sstable-(5) : 1</tspan></text><rect x="49" y="3811.2495" width="70.75" height="18.500217" fill="#0a2832"/><rect x="49" y="3811.2495" width="70.75" height="18.500217" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(49 3816.2074)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="3.869873" y="7" textLength="63.010254">sstable-(1,2,3)</tspan></text><rect x="125.375" y="3811.2496" width="70.75" height="18.500217" fill="#0a2832"/><rect x="125.375" y="3811.2496" width="70.75" height="18.500217" stroke="black" stroke-linecap=" round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(125.375 3816.2076)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="3.869873" y="7" textLength="63.010254">sstable-(4,5,6)</tspan></text><text transform="translate(11.5 3813.4997)" fill="#a57706"><tspan font-family="Helvetica" font-size="12" font-weight="bold" fill="#a57706" x=".26660156" y="11" textLength="16.669922">CP</tspan><tspan font-family="Helvetica" font-size="12" font-weight="bold" fill="#a57706" x="16.725586" y="11" textLength="10.0078125"> 4</tspan></text><text transform="translate(166 3752.5619)" fill="black"><tspan font-family="Helvetica" font-size="9" font-weight="500" x=".24487305" y="9" textLength="25.510254">merge</tspan></text><rect x="360.65693" y="3794" width="207.84308" height="33.5001" fill="black"/><rect x="360.65693" y="3794" width="207.84308" height="33.5001" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><tex t transform="translate(365.65693 3798.9578)" fill="#2076c8"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#2076c8" x="0" y="7" textLength="197.43213">op-2-1-sstable-(1,2,3) -> (dfs://cp-3/38-46-83)</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="#a57706" x="0" y="21" textLength="189.03076">op-2-1-sstable-(4,5,6) -> dfs://cp-3/38-46-83</tspan></text><rect x="579.28193" y="3794" width="122.15423" height="54.743706" fill="black"/><rect x="579.28193" y="3794" width="122.15423" height="54.743706" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(584.28193 3795.5796)" fill="white"><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="7" textLength="109.217773">op-2-1-sstable-(1,2,3) : 2</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="21" textLength="92.41504">op-2-1-sstable-(4) : 1</tspan><tspan font-family="Monaco " font-size="7" font-weight="500" fill="white" x="0" y="35" textLength="92.41504">op-2-1-sstable-(5) : 1</tspan><tspan font-family="Monaco" font-size="7" font-weight="500" fill="white" x="0" y="49" textLength="109.217773">op-2-1-sstable-(4,5,6) : 1</tspan></text><line x1="11" y1="3620.75" x2="788" y2="3620.75" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1" stroke-dasharray="1,4"/><line x1="11" y1="3697.25" x2="788" y2="3697.25" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1" stroke-dasharray="1,4"/><line x1="11" y1="3788.75" x2="788" y2="3788.75" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1" stroke-dasharray="1,4"/><line x1="717.2181" y1="3589.599" x2="717.2181" y2="3609.6928" marker-end="url(#DimensionArrow_Marker)" marker-start="url(#DimensionArrow_Marker_2)" stroke="#738a05" stroke-linecap="round" stroke-linejoin="round" stroke-width="2"/><line x1="738.25" y1="3589.5993" x2="738.25" y2 ="3685.65" marker-end="url(#DimensionArrow_Marker_3)" marker-start="url(#DimensionArrow_Marker_4)" stroke="#bd3612" stroke-linecap="round" stroke-linejoin="round" stroke-width="2"/><line x1="759.16273" y1="3631.8496" x2="757.83727" y2="3777.6504" marker-end="url(#DimensionArrow_Marker_5)" marker-start="url(#DimensionArrow_Marker_6)" stroke="#2076c8" stroke-linecap="round" stroke-linejoin="round" stroke-width="2"/><line x1="779.5" y1="3708.35" x2="779.5" y2="3840.1437" marker-end="url(#DimensionArrow_Marker_7)" marker-start="url(#DimensionArrow_Marker_8)" stroke="#a57706" stroke-linecap="round" stroke-linejoin="round" stroke-width="2"/><text transform="translate(121 3547)" fill="black"><tspan font-family="Helvetica" font-size="12" font-weight="bold" x="2.1445312" y="11" textLength="146.04492">Local RocksDB directory </tspan><tspan font-family="Helvetica" font-size="12" font-weight="bold" x="1.8134766" y="25" textLength="143.37305">for Operator 2, Subtask 1</tspan></text><text transfo rm="translate(383.07846 3547.0001)" fill="black"><tspan font-family="Helvetica" font-size="12" font-weight="bold" x=".1484375" y="11" textLength="162.703125">Shared States in Checkpoint</tspan><tspan font-family="Helvetica" font-size="12" font-weight="bold" x="6.9921875" y="25" textLength="135.01172">(Key -> uploaded DFS fi</tspan><tspan font-family="Helvetica" font-size="12" font-weight="bold" x="142.00391" y="25" textLength="14.003906">le)</tspan></text><text transform="translate(593.125 3547.0003)" fill="black"><tspan font-family="Helvetica" font-size="12" font-weight="bold" x="9.8203125" y="11" textLength="76.69336">Shared State </tspan><tspan font-family="Helvetica" font-size="12" font-weight="bold" x=".16113281" y="25" textLength="92.677734">Registry Counts</tspan></text><text transform="translate(712 3547.0003)" fill="black"><tspan font-family="Helvetica" font-size="12" font-weight="bold" x="11.161133" y="11" textLength="54.01172">Retained </tspan><tspan font-family="Helve tica" font-size="12" font-weight="bold" x=".49414062" y="25" textLength="72.01172">Checkpoints</tspan></text><line x1="42" y1="3550.5" x2="42" y2="3850.7437" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1" stroke-dasharray="1,4"/><line x1="354.5" y1="3550.5" x2="354.5" y2="3849.7437" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1" stroke-dasharray="1,4"/><line x1="574" y1="3550.5" x2="574" y2="3848.7437" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1" stroke-dasharray="1,4"/><line x1="706.875" y1="3550.5" x2="706.875" y2="3848.7437" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1" stroke-dasharray="1,4"/><line x1="12" y1="3578.2088" x2="789" y2="3578.2088" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1" stroke-dasharray="1,4"/><text transform="translate(191 3796.4586)" fill="black"><tspan font-family="Monaco" font-size="7" font-wei ght="500" fill="black" x=".29589844" y="7" textLength="50.408203">+sstable-(6)</tspan></text></g></g></svg>
