metorikku: a YAML-driven ETL framework on top of Apache Spark

Two files describe the whole pipeline

Metorikku is a library that simplifies writing and executing ETLs on top of Apache Spark, based on simple YAML configuration files, and it runs on any Spark cluster. To use it you define two files. The metric file defines the steps and queries of the ETL as well as where and what to output, and the job file includes the input sources, the output destinations and the location of the metric configuration files. JSON is supported in place of YAML for both. The split is the design decision worth understanding. A metric is a transformation, with inputs and outputs named independently of where the data physically lives, and a job is one run of a metric against a particular set of sources. The same metric can therefore run in development against small mocked inputs and in production against Kafka, without the metric changing.

A metric file is a list of named dataframes and the SQL between them

The metric file format is short enough to read in one screen.

steps:
- dataFrameName: df1
  checkpoint: true #This persists the dataframe to storage and truncates the execution plan. For more details, see https://jaceklaskowski.gitbooks.io/mastering-spark-sql/content/spark-sql-checkpointing.html
  sql:
    SELECT *
    FROM input_1
    WHERE id > 100
- dataFrameName: df2
  sql:
    SELECT *
    FROM df1
    WHERE id < 1000
output:
- dataFrameName: df2
  outputType: Parquet
  outputOptions:
    saveMode: Overwrite
    path: df2.parquet

Each step names a dataframe and gives the SQL that produces it, and later steps refer to earlier dataframes by name, so df2 selects from df1. The checkpoint option persists the dataframe to storage and truncates the execution plan, which is the usual Spark remedy for a plan that grows too large. The output section names a dataframe, an output type and the options for it, here Parquet with overwrite mode and a path. A full sample configuration with every possible value is in the repository's config directory, and the README points at the Spark SQL language manual for what you can put in the sql blocks.

The job file names where data comes from and where it goes

The job file is the smaller of the two and wires the metric to the outside world.

metrics:
  - /full/path/to/your/metric/file.yaml
inputs:
  input_1: parquet/input_1.parquet
  input_2: parquet/input_2.parquet
output:
    file:
        dir: /path/to/parquet/output

The metrics entry is a path to a metric file, which may be a full path or a Hadoop-compatible location such as s3 or hdfs. The inputs map gives a name to a location, and those names are what the metric's SQL refers to, so input_1 in the metric is the source the job file maps. The output section says where the results land. The README also documents a runtime environment variable, CONFIG_FILES_PATH_PREFIX, that adds a prefix to all configuration file paths, which is how you run the same job file locally and remotely without editing it, and there is a sample job configuration in the config directory and a set of worked examples in the examples directory.

Connectors, and a mismatch between inputs and outputs worth noticing

The supported inputs are CSV, JSON, Parquet, JDBC, Kafka, Cassandra and Elasticsearch. The supported outputs are CSV, JSON, Parquet, Redshift, Cassandra, Segment, JDBC, Kafka and Elasticsearch. Read the two lists side by side, because the asymmetry is the useful detail. Kafka and Cassandra are inputs but not outputs, Redshift and Segment are outputs but not inputs, and Elasticsearch appears on both sides. So a pipeline can read from a Kafka topic and write into a Redshift table, but it cannot write to Kafka in this version, which is a constraint you would hit immediately if you were planning a streaming-to-streaming path. The framework also adds a small set of extras beyond connectors, notably a regular expression matching scalar function, and the examples directory shows that it handles file date range inputs, file input streams, and an atlas example, plus a data quality example, an Elasticsearch example, a Hive example, a Hudi example, an InfluxDB example, a Kafka example and a user defined function example, which is a good sign about the range the author tested.

The tester, and why the key columns exist

The Metorikku Tester exists in order to test and fully automate the deployment of metrics, and it is the part of the project with no real equivalent in the Spark ecosystem. A test is made of a test settings file and mocked data. The settings define which metric to test and where to get the mock data, and all the paths must be relative to the directory of the test file.

metric: "/path/to/metric"
mocks:
- name: table_1
  path: mocks/table_1.jsonl
tests:
  df2:
  - id: 200
    name: test
  - id: 300
    name: test2
keys:
  df2:
  - id
  - name

The mocks are the expected rows as newline-delimited JSON.

{ "id": 200, "name": "test" }
{ "id": 300, "name": "test2" }
{ "id": 1, "name": "test3" }

The keys section defines the unique columns of each expected dataframe result, and every expected row must have a unique combination of the key values. It is optional, and for a dataframe without a key definition all the columns of the first expected row are used as the unique keys by default. A shorter key list gives better performance and a more detailed error message when a test fails, and the structure of the expected result must be identical for all rows, with keys valid as columns of the same dataframe's expected result.

Streaming metrics need a different mock, and a different output mode

The tester also handles the case where the real pipeline is a streaming one, because Spark behaves differently when writing queries against streaming sources such as Kafka, and the README wants the test to behave like the real thing. You configure a mock to behave like a streaming input, and the output mode changes as a result.

metric: "/path/to/metric"
mocks:
- name: table_1
  path: mocks/table_1.jsonl
  # default is false
  streaming: true
# default is append output mode
outputMode: update
tests:
  df2:
  - id: 200
    name: test
  - id: 300
    name: test2

The tester runs the same way as Metorikku itself, with one difference, that the main class changes from com.yotpo.metorikku.Metorikku to com.yotpo.metorikku.MetorikkuTester. That is a small detail with a real consequence. A metric that reads from Kafka has semantics that only appear in streaming execution, so a test that runs in batch mode can pass while the production run behaves differently, and the streaming mock plus the update output mode is the mechanism for closing that gap.

Three ways to run it, and a version story that dates the project

There are three options to run Metorikku. On a Spark cluster it requires Apache Spark 2.2 or later, you download the last released jar, and you submit it with a class and a config file. Remote job and metric files work too, since you can write a full path such as an s3 location and anything supported by Hadoop can be used. Locally there is a standalone jar with a bundled Spark, run with a local master setting and the main class, and a job in JSON form is supported with a different argument form. The README also documents how to run it inside IntelliJ, listing the main class, the VM options with the local master, an executor core count, a driver bind address and the Kryo serializer, plus the program arguments, and how to run the tester from the IDE with its own main class. The version story is where the caution goes. The library requires Scala 2.11 for Spark 2 or 2.12 for Spark 3, the local run path is documented as needing Java 1.8, and the last release tags are v0.0.156 from February 2023 with the two before it in 2022, even though the last push was on 2026-09-01. So the source is being touched but the releases are old, and the dependency coordinates in a build file are given as the library name with the version left as a placeholder, which is a sign you are expected to pin a version yourself.

Editorial conclusion

metorikku fits a data team that wants Spark SQL pipelines defined in version-controlled YAML rather than Scala, and that wants to unit test a metric before deploying it, since the tester is the part almost no Spark tool offers. It does not fit a team that has moved to Spark 3 with a modern Scala toolchain, since the library requires Scala 2.11 for Spark 2 or 2.12 for Spark 3, and the local run path expects Java 1.8. Verify first which Spark and Scala versions your cluster runs, read the sample configuration files rather than starting from an empty file, and note that the release tags are old, so you are adopting a project whose recent commit history matters more than its version number.

Frequently asked questions

Official sources

1. License: MIT
2. Project website
3. README
4. Releases
5. YotpoLtd/metorikku on GitHub