# OpenRefine on Spark prototype

This document describes the prototype developed as a proof of concept to migrate OpenRefine away from its in-memory data processing backend to an architecture where project data does not need to be loaded in memory fully.

## Introduction

Changing the data processing backend requires a complete overhaul of OpenRefine's architecture. Our goal is to show that this can be done with little changes to the user experience for small datasets, preserving the existing use cases for the tool.
We do not aim to preserve the Java API exposed to extensions nor the REST API exposed
to the web UI, although the changes made to the latter are minimal.

This document proposes a new architecture for OpenRefine's backend, which uses Spark to implement all steps in a data cleaning workflow. Spark offers three API layers to manipulate datasets:
* *Resilient Distributed Datasets* (RDD), which are the lowest API level. RDDs are sets of data records chunked into partitions, which can be distributed on different nodes of a cluster. The Spark Java API provides views on these RDDs, with methods letting the user derive other RDDs by transformation (which are computed lazily) or materialized in-memory results (via actions).
* *Datasets*, which provide a more efficient serialization of the Java objects stored as records, rather than using the default Java serialization (used in RDDs). This requires declaring explicit encoders for all classes used in datasets.
* *Dataframes*, which add a tabular structure to datasets. As of Spark 2.x this restricts the datatypes that can be stored as values, since support for User Defined Types (UDTs) has been made private and [might not be reintroduced soon](https://github.com/apache/spark/pull/16478).

The current prototype uses Spark at the lowest API level: RDDs. This choice is motivated by the following observations:
* OpenRefine lets users store any `Serializable` object as cell value, and cells can also contain reconciliation information. This makes it impossible to use the Dataframe API (since it would require user defined types). Using the Dataset API would only be possible if we start restricting the types of values stored in cells or devise a mechanism to register encoders for custom values.
* The RDD API provides an indexed view on RDDs which lets us control the indexing of the rows by row id, which does not seem to have an equivalent at the Dataset level.
* The RDD API can be used via Apache Beam whereas [support for the higher levels is experimental](https://beam.apache.org/documentation/runners/spark/).

When (if?) Spark adds back public support for User Defined Types, we might consider migrating to the Dataframes API.

## What is in the prototype

A large part of OpenRefine's backend architecture relies on the expectation that all project data is available in memory. Changing this requires adaptations to:
* the core data model (Project, History, Change, Row, Column…)
* all facets and related logic (44 classes)
* all operations (35 classes)
* all changes (20 classes)
* all importers (12 classes)
* all exporters (10 classes)

This prototype shows how this migration can be done for the core datamodel and selected classes, for demonstration purposes. It supports only:
* one facet type (text facets)
* one operation type (mass edit, obtained by editing a value in a text facet)
* two change types: mass edit changes and single cell edits
* one importer: CSV/TSV
* no exporter (since their implementation is not expected to be challenging)

The prototype only implements the rows mode. We plan to implement the records mode in a naive way: all project data will be loaded in memory for that. This means that the records mode should remain usable in its current form for small datasets. Implementing a scalable records mode will require considerable changes to its current design and is currently being discussed [on the mailing list](https://groups.google.com/forum/#!msg/openrefine/X9O8NBC1UKQ/alxV_f3BBwAJ) and related issues (linked from the thread).

## Trying out the prototype

1. Build the prototype from the [`spark-prototype` branch](https://github.com/OpenRefine/OpenRefine/tree/spark-prototype) with `./refine clean` and `./refine build`
2. Run OpenRefine, preferably on a clean workspace: `./refine -d /tmp/openrefine-spark`. Running it on your existing workspace will display existing projects as corrupt, and any projects you create with the prototype will not be readable by previous versions.
3. Import a simple project for instance by typing in a small CSV project in the Clipboard field
4. Play around by adding facets, editing facet values, and editing single cells.

You can get a sense of which Spark jobs are submitted by looking at the Spark web UI
while OpenRefine is running: [http://localhost:4040/stages/](http://localhost:4040/stages/). (It might be on a different port on your computer: check the OpenRefine logs for that).

## Previous architecture

In OpenRefine 3.x, project data is held entirely in memory: the [`Project`](https://github.com/OpenRefine/OpenRefine/blob/master/main/src/com/google/refine/model/Project.java#L63) class has
[a `rows` field](https://github.com/OpenRefine/OpenRefine/blob/master/main/src/com/google/refine/model/Project.java#L68) which contains the list of rows of the project.

Each operation generates a [`Change`](https://github.com/OpenRefine/OpenRefine/blob/master/main/src/com/google/refine/history/Change.java#L49) which with two methods:
* `apply`, which replaces the `Project` data in place by the new data introduced by the change;
* `revert`, which reverts the project data back to its original state, before the operation is applied.

Changes are required not to perform any computation themselves: this means they store the old and new values of the cells they change. All changes in a project are serialized on disk in a custom text-based format (which is defined independently by each change type).

To save memory, change data is only loaded when a change needs to be applied or reverted: this means that loading a project with hundreds of changes only amounts to loading the table in the state it was left at. However, reverting back to the initial project state is an expensive operation, and prone to failures since if a single change fails to revert (or does not revert correctly) this can have cascading consequences (and data loss).

This architecture fundamentally relies on the fact that all project data is loaded in memory.

Facet statistics are computed [using a Visitor pattern](https://github.com/OpenRefine/OpenRefine/blob/master/main/src/com/google/refine/browsing/FilteredRows.java#L42) which visits a selection of rows defined by other facets. This stateful pattern does not parallelize well since it requires the visitor to hold in its state the statistics of the rows seen so far. In addition, the implementation does one pass over the entire data for each facet.

## Proposed new architecture

We remove the `rows` attribute in the `Project` class. Instead, we introduce a new [`GridState`](https://github.com/OpenRefine/OpenRefine/blob/spark-prototype/or-model/src/main/java/org/openrefine/model/GridState.java#L35) class which holds the project data at a particular point in the history (column headers, cell data and overlay models). Because cell data is held as a Spark RDD, a `GridState` only contains a recipe to compute the state of the project at this specific point, from the initial project data. This has important consequences:
* We can afford to store `GridState` objects for all transformation steps in the history (whereas in the previous architecture only the current state of the project was kept);
* Therefore, reverting back to a previous state (with the undo command) only amounts to changing the position in the history, which is instantaneous. In the previous architecture, undoing many changes at once could take quite some time;
* The [`Change`](https://github.com/OpenRefine/OpenRefine/blob/spark-prototype/or-model/src/main/java/org/openrefine/history/Change.java#L58) classes only need to know how to apply a transformation, not how to revert it. This simplifies the code a lot, and avoids the duplication of stored data.

Here is an overview of the object structure for a project with three operations applied:


This new architecture means that we can also significantly simplify the way cell data is stored:
* The system of cell indices has been removed. The index of a cell in a row object is simply the index of its column in the list of columns.
* Storing reconciliation information separately (in a `Pool`) is no longer required.

The architecture of [facets](https://github.com/OpenRefine/OpenRefine/blob/spark-prototype/or-model/src/main/java/org/openrefine/browsing/facets/Facet.java#L43) has also been significantly changed. The state of all facets is computed on the current `GridState` by computing an aggregation on the underlying RDD. This means that all facets must:
* Provide an empty state (`FacetState`) which represents the facets statistics on an empty table;
* Provide an aggregator ([`FacetAggregator`](https://github.com/OpenRefine/OpenRefine/blob/spark-prototype/or-model/src/main/java/org/openrefine/browsing/facets/FacetAggregator.java#L18)) which knows how to add a row to a `FacetState`, merge two `FacetState`s representing statistics obtained from different partitions of rows, and determine if a row is selected by the facet or not.

With this architecture, this means that the state of all facets can be computed [in
a single aggregation over the entire grid](https://github.com/OpenRefine/OpenRefine/blob/spark-prototype/or-model/src/main/java/org/openrefine/browsing/facets/AllFacetsAggregator.java#L17). This architecture should be easy to adapt later on if we want to add support for approximate facets which only compute their statistics on a sample of rows.

## Performance optimizations planned

As this is just a prototype, no optimizations have been done at all. Here are a few that could be done:

### Project serialization

Project serialization is currently quite naive, it just uses Java serialization which is quite verbose. We expect that it should be possible to improve this by using a dedicated serializer.

### Better use of the RDD API

The implementation of OpenRefine operations using the RDD API can be optimized to give simpler Directed Acyclic Graphs of operations, speeding up the computations.
This optimization can be done relatively easily via Spark's web UI.

### Importer implementation

Currently, the CSV/TSV importer implemented reads all the data in RAM, and then sends it to Spark. This is obviously a bottleneck that we should get rid of in order to deal with large input files.

This requires [implementing importers by following Hadoop's API](https://www.ae.be/blog-en/ingesting-data-spark-using-custom-hadoop-fileinputformat/), to create partitions of the dataset on the fly directly when reading the data. This is not going to be possible for all input formats as it requires to efficiently detect the number of rows and read selected row chunks. For instance, ODF and XLS importers are generally used for small datasets as the software that produce them are not designed to deal with large datasets (or in some cases the formats have explicit caps on the number of rows). Those formats will remain on the existing architecture which loads all the data in memory at import time.

### Action optimizations

The results of some actions, such as computing the number of rows in a grid state, could be cached in the GridState class to avoid recomputations.

### Facet computations

Facet states could potentially benefit with some caching as well. For large datasets we anticipate that we should introduce sampling to compute facets: we would let the user define on how many rows to compute the facet statistics. This can then ensure that facets can be refreshed in real time.

### Persisting processing stages

As the history grows longer it might be necessary to persist strategic history stages to avoid recomputing them often. Some logic for this is already present in Spark so we need to experiment to check if we need to implement our own logic on top of it.