Dataflow Runner

Dataflow Runner is a system for creating and running AWS EMR jobflow clusters and steps. It uses templated playbooks to define your cluster, and the Hadoop/Spark/et al jobs that you want to run.

Installation

Platform native binaries are available from the GitHub releases page.
Docker images are available at DockerHub as of version 0.7.3.

Cluster Configuration

A cluster configuration contains all of the information needed to create a new cluster which is ready to accept a playbook. Currently AWS EMR is the only supported data-flow fabric.

For the cluster template see: config/cluster.json.sample

Playbook Configuration

A playbook consists of one of more steps. Steps are added to the cluster and run in series.

For the playbook template see: config/playbook.json.sample

Templates

Configuration files are run through Golang’s text template processor. The template processor can access all variables defined on the command line using the --vars argument.

For example to use the --vars argument with a playbook step:

{
  "type": "CUSTOM_JAR",
  "name": "Combine Months",
  "actionOnFailure": "CANCEL_AND_WAIT",
  "jar": "s3://snowplow-hosted-assets/3-enrich/hadoop-event-recovery/snowplow-hadoop-event-recovery-0.2.0.jar",
  "arguments": [
    "com.snowplowanalytics.hadoop.scalding.SnowplowEventRecoveryJob",
    "--hdfs",
    "--input",
    "hdfs:///local/monthly/{{.inputVariable}}",
    "--output",
    "hdfs:///local/recovery/{{.outputVariable}}"
  ]
}

You would then pass the following command:

host> ./dataflow-runner run --emr-playbook ${emr-playbook-path} --emr-cluster j-2DPBXD87LSGP9 --vars inputVariable,input,outputVariable,output

This would resolve to:

{
  "type": "CUSTOM_JAR",
  "name": "Combine Months",
  "actionOnFailure": "CANCEL_AND_WAIT",
  "jar": "s3://snowplow-hosted-assets/3-enrich/hadoop-event-recovery/snowplow-hadoop-event-recovery-0.2.0.jar",
  "arguments": [
    "com.snowplowanalytics.hadoop.scalding.SnowplowEventRecoveryJob",
    "--hdfs",
    "--input",
    "hdfs:///local/monthly/input",
    "--output",
    "hdfs:///local/recovery/output"
  ]
}

The following custom functions are also supported:

nowWithFormat [timeFormat]: where timeFormat is a valid Golang time format
timeWithFormat [epoch] [timeFormat]: where epoch is the number of seconds elapsed between January 1st 1970 and a certain point in time as a string and timeFormat is valid Golang time format
systemEnv "ENV_VAR": where ENV_VAR is a key for a valid environment variable
base64 [string]: will base64-encode the string passed as argument
base64File "path/to/file.txt": will base64-encode the content of the file located at the path passed as argument

CLI Commands

There are several commands that can be used to manage your data-flow fabric.

`up`: Launches a new EMR cluster

NAME:
   dataflow-runner up - Launches a new EMR cluster

USAGE:
   dataflow-runner up [command options] [arguments...]

OPTIONS:
   --emr-config value  EMR config path
   --vars value        Variables that will be used by the templater

This command will launch a new cluster ready for step execution, the output should look something like the following:

NAME:
   dataflow-runner run - Adds jobflow steps to a running EMR cluster

USAGE:
   dataflow-runner run [command options] [arguments...]

OPTIONS:
   --emr-playbook value  Playbook path
   --emr-cluster value   Jobflow ID
   --async               Asynchronous execution of the jobflow steps
   --lock value          Path to the lock held for the duration of the jobflow steps. This is materialized by a file or a KV entry in Consul depending on the --consul flag.
   --softLock value      Path to the lock held for the duration of the jobflow steps. This is materialized by a file or a KV entry in Consul depending on the --consul flag. Released no matter if the operation failed or succeeded.
   --consul value        Address of the Consul server used for distributed locking
   --vars value          Variables that will be used by the templater

`run`: Adds jobflow steps to a running EMR cluster

host> ./dataflow-runner up --emr-config ${emr-config-path}
INFO[0001] Launching EMR cluster with name 'dataflow-runner - sample name'...
INFO[0001] EMR cluster is in state STARTING - need state WAITING, checking again in 20 seconds...
INFO[0021] EMR cluster is in state STARTING - need state WAITING, checking again in 20 seconds...
# this goes for a few lines, omitted for brevity
INFO[0227] EMR cluster is in state STARTING - need state WAITING, checking again in 20 seconds...
INFO[0248] EMR cluster is in state BOOTSTRAPPING - need state WAITING, checking again in 20 seconds...
INFO[0269] EMR cluster is in state BOOTSTRAPPING - need state WAITING, checking again in 20 seconds...
INFO[0289] EMR cluster is in state BOOTSTRAPPING - need state WAITING, checking again in 20 seconds...
INFO[0310] EMR cluster launched successfully; Jobflow ID: j-2DPBXD87LSGP9

This command adds new steps to the already running cluster. By default this command is blocking - however if you wish to submit and forget simply supply the --async argument, the output should look something like the following:

host> ./dataflow-runner run --emr-playbook ${emr-playbook-path} --emr-cluster j-2DPBXD87LSGP9
INFO[0310] Successfully added 2 steps to the EMR cluster with jobflow id 'j-2DPBXD87LSGP9'...
ERRO[0357] Step 'Combine Months' with id 's-9WZ0VFKC770J' was FAILED
ERRO[0358] Step 'Combine Months 2' with id 's-37F9PKSXBHDAU' was CANCELLED
ERRO[0358] 2/2 steps failed to complete successfully

In this case the first step failed which meant that the second step was cancelled. This behavior is dependent on your actionOnFailure - you can choose either to:

“CANCEL_AND_WAIT”: This will cancel all other currently queued jobs and return the cluster to a waiting state ready for new job submissions.
“CONTINUE”: This will go to the next step regardless if it failed or not.

Note: We have removed the ability to terminate the job flow on failure, to terminate you will need to use the down command.

Additionally, Dataflow Runner can acquire a lock before starting the job which can prevent other jobs from running at the same time. Its release will happen when:

the job has terminated (whether successfully or with failure) with the --softLock flag
the job has succeeded with the --lock flag (“hard lock”)

As the above implies, if a job were to fail and the --lock flag was used, manual cleaning of the lock will be required.

Additionally, supplying a Consul address, through the --consul flag will make this lock distributed.

When the --consul flag is used, the lock will be materialized by a key-value pair in Consul for which the key is the value supplied with the --lock or --softLock argument. Otherwise, it will be materialized by a file on the machine located at the specified path (either relative to your working directory or absolute).

`down`: Terminates a running EMR cluster

NAME:
   dataflow-runner down - Terminates a running EMR cluster

USAGE:
   dataflow-runner down [command options] [arguments...]

OPTIONS:
   --emr-config value   EMR config path
   --emr-cluster value  Jobflow ID
   --vars value         Variables that will be used by the templater

When you are done with the EMR cluster you can terminate it by using the down command. This takes the original emr configuration and the job flow id to then go and terminate the cluster, the output should look something like the following:

host> ./dataflow-runner down --emr-config ${emr-config-path} --emr-cluster j-2DPBXD87LSGP9
INFO[0358] Terminating EMR cluster with jobflow id 'j-2DPBXD87LSGP9'...
INFO[0358] EMR cluster is in state TERMINATING - need state TERMINATED, checking again in 20 seconds...
INFO[0378] EMR cluster is in state TERMINATING - need state TERMINATED, checking again in 20 seconds...
INFO[0399] EMR cluster is in state TERMINATING - need state TERMINATED, checking again in 20 seconds...
INFO[0420] EMR cluster is in state TERMINATING - need state TERMINATED, checking again in 20 seconds...
INFO[0440] Transient EMR run completed successfully

`run-transient`: Launches, runs and then terminates an EMR cluster

NAME:
   dataflow-runner run-transient - Launches, runs and then terminates an EMR cluster

USAGE:
   dataflow-runner run-transient [command options] [arguments...]

OPTIONS:
   --emr-config value    EMR config path
   --emr-playbook value  Playbook path
   --lock value          Path to the lock held for the duration of the jobflow steps. This is materialized by a file or a KV entry in Consul depending on the --consul flag.
   --softLock value      Path to the lock held for the duration of the jobflow steps. This is materialized by a file or a KV entry in Consul depending on the --consul flag. Released no matter if the operation failed or succeeded.
   --consul value        Address of the Consul server used for distributed locking
   --vars value          Variables that will be used by the templater

This command is a combination of up, run and down which is designed to mimic the current EmrEtlRunner behavior.

Installation​

Cluster Configuration​

Playbook Configuration​

Templates​

CLI Commands​

up: Launches a new EMR cluster​

run: Adds jobflow steps to a running EMR cluster​

down: Terminates a running EMR cluster​

run-transient: Launches, runs and then terminates an EMR cluster​