Metorikku is a library that simplifies writing and executing ETLs on top of Apache Spark.

It is based on simple YAML configuration files and runs on any Spark cluster.

The platform also includes a simple way to write unit and E2E tests.

Getting started

To run Metorikku you must first define 2 files.

Metric file

A metric file defines the steps and queries of the ETL as well as where and what to output.

For example a simple configuration YAML (JSON is also supported) should be as follows:

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

You can check out a full example file for all possible values in the sample YAML configuration file.

Make sure to also check out the full Spark SQL Language manual for the possible queries.

Job file

This file will include input sources, output destinations and the location of the metric config files.

So for example a simple YAML (JSON is also supported) should be as follows:

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

You can check out a full example file for all possible values in the sample YAML configuration file.

Also make sure to check out all our examples.

Supported input/output:

Currently Metorikku supports the following inputs: CSV, JSON, parquet, JDBC, Kafka, Cassandra, Elasticsearch

And the following outputs: CSV, JSON, parquet, Redshift, Cassandra, Segment, JDBC, Kafka, Elasticsearch

Running Metorikku

There are currently 3 options to run Metorikku.

Run on a spark cluster

To run on a cluster Metorikku requires Apache Spark v2.2+

• Download the last released JAR
• Run the following command: spark-submit --class com.yotpo.metorikku.Metorikku metorikku.jar -c config.yaml

Running with remote job/metric files:

Metorikku supports using remote job/metric files.

Simply write the full path to the job/metric. example: s3://bucket/job.yaml

Anything supported by hadoop can be used (s3, hdfs etc.)

To help running both locally and remotely you can add the following env variable at runtime to add a prefix to all your configuration files paths: CONFIG_FILES_PATH_PREFIX=s3://bucket/

Run locally

Metorikku is released with a JAR that includes a bundled spark.

• Download the last released Standalone JAR
• Metorikku is required to be running with Java 1.8
• Run the following command: java -D"spark.master=local[*]" -cp metorikku-standalone.jar com.yotpo.metorikku.Metorikku -c config.yaml
• Also job in a JSON format is supported, run following command: java -D"spark.master=local[*]" -cp metorikku-standalone.jar com.yotpo.metorikku.Metorikku --job "{*}"

Run locally in intellij:

Go to Run->Edit Configuration->add application configuration

• Main Class: com.yotpo.metorikku.Metorikku
• Vm options: -Dspark.master=local[*] -Dspark.executor.cores=1 -Dspark.driver.bindAddress=127.0.0.1 -Dspark.serializer=org.apache.spark.serializer.KryoSerializer
• program arguments: -c examples/movies.yaml
• JRE: 1.8

Run tester in intellij:

• Main class: com.yotpo.metorikku.MetorikkuTester
• Program arguments: --test-settings /{path to }/test_settings.yaml

Run as a library

It's also possible to use Metorikku inside your own software

Metorikku library requires scala 2.11 (spark 2)/2.12 (spark 3)

To use it add the following dependency to your build.sbt: "com.yotpo" % "metorikku" % "LATEST VERSION"

Metorikku Tester

In order to test and fully automate the deployment of metrics we added a method to run tests against a metric.

A test is comprised of the following:

Test settings

This defines what to test and where to get the mocked data.

** All the paths must be relative to the directory of the test file. **

For example, a simple test YAML (JSON is also supported) will be:

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

And the corresponding mocks/table_1.jsonl:

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

The Keys section allows the user to define the unique columns of every DataFrame's expected results - every expected row result should have a unique combination for the values of the key columns. This part is optional and can be used to define only part of the expected DataFrames - for the DataFrames that don't have a key definition, all of the columns defined in the first row result will be taken by default as the unique keys. Defining a shorter list of key columns will result in better performances and a more detailed error message in case of test failure.

The structure of the defined expected dataFrame's result must be identical for all rows, and the keys must be valid (defined as columns of the expected results of the same DataFrame)

Running Metorikku Tester

You can run Metorikku tester in any of the above methods (just like a normal Metorikku).

The main class changes from com.yotpo.metorikku.Metorikku to com.yotpo.metorikku.MetorikkuTester

Testing streaming metrics

In Spark some behaviors are different when writing queries for streaming sources (for example kafka).

In order to make sure the test behaves the same as the real life queries, you can configure a mock to behave like a streaming input by writing the following:

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

Notes

Variable interpolation

All configuration files support variable interpolation from environment variables and system properties using the following format: ${variable_name}

Using JDBC

When using JDBC writer or input you must provide a path to the driver JAR.

For example to run with spark-submit with a mysql driver: spark-submit --driver-class-path mysql-connector-java-5.1.45.jar --jars mysql-connector-java-5.1.45.jar --class com.yotpo.metorikku.Metorikku metorikku.jar -c config.yaml

If you want to run this with the standalone JAR: java -Dspark.master=local[*] -cp metorikku-standalone.jar:mysql-connector-java-5.1.45.jar -c config.yaml

JDBC query

JDBC query output allows running a query for each record in the dataframe.

Mandatory parameters:

• query - defines the SQL query. In the query you can address the column of the DataFrame by their location using the dollar sign ($) followed by the column index. For example:

INSERT INTO table_name (column1, column2, column3, ...) VALUES ($1, $2, $3, ...);

Optional Parameters:

• maxBatchSize - The maximum size of queries to execute against the DB in one commit.
• minPartitions - Minimum partitions in the DataFrame - may cause repartition.
• maxPartitions - Maximum partitions in the DataFrame - may cause coalesce.

Kafka output

Kafka output allows writing batch operations to kafka

We use spark-sql-kafka-0-10 as a provided jar - spark-submit command should look like so:

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 --class com.yotpo.metorikku.Metorikku metorikku.jar

Mandatory parameters:

• topic - defines the topic in kafka which the data will be written to. currently supported only one topic

• valueColumn - defines the values which will be written to the Kafka topic, Usually a json version of data, For example:

SELECT keyColumn, to_json(struct(*)) AS valueColumn FROM table

Optional Parameters:

• keyColumn - key that can be used to perform de-duplication when re