Data pipeline asset management with Dataflow | by Netflix Technology Blog

by Sam Setegne, Jai Balani, Olek Gorajek

• asset — any enterprise logic code in a uncooked (e.g. SQL) or compiled (e.g. JAR) kind to be executed as a part of the person outlined knowledge pipeline.
• knowledge pipeline — a set of duties (or jobs) to be executed in a predefined order (a.okay.a. DAG) for the aim of remodeling knowledge utilizing some enterprise logic.
• Dataflow — Netflix homegrown CLI instrument for knowledge pipeline administration.
• job — a.okay.a process, an atomic unit of knowledge transformation logic, a non-separable execution block within the workflow chain.
• namespace — distinctive label, normally representing a enterprise topic space, assigned to a workflow asset to establish it throughout all different belongings managed by Dataflow (e.g. safety).
• workflow — see “knowledge pipeline”

The issue of managing scheduled workflows and their belongings is as outdated as using cron daemon in early Unix working programs. The design of a cron job is straightforward, you are taking some system command, you choose the schedule to run it on and you might be executed. Instance:

0 0 * * MON /dwelling/alice/backup.sh

Within the above instance the system would get up each Monday morning and execute the backup.sh script. Easy proper? However what if the script doesn’t exist within the given path, or what if it existed initially however then Alice let Bob entry her dwelling listing and he unintentionally deleted it? Or what if Alice wished so as to add new backup performance and she or he unintentionally broke present code whereas updating it?

The solutions to those questions is one thing we want to handle on this article and suggest a clear resolution to this downside.

Let’s outline some necessities that we’re serious about delivering to the Netflix knowledge engineers or anybody who want to schedule a workflow with some exterior belongings in it. By exterior belongings we merely imply some executable carrying the precise enterprise logic of the job. It may very well be a JAR compiled from Scala, a Python script or module, or a easy SQL file. The essential factor is that this enterprise logic might be in-built a separate repository and maintained independently from the workflow definition. Preserving all that in thoughts we want to obtain the next properties for the entire workflow deployment:

1. Versioning: we wish each the workflow definition and its belongings to be versioned and we wish the variations to be tied collectively in a transparent approach.
2. Transparency: we need to know which model of an asset is working together with each workflow occasion, so if there are any points we are able to simply establish which model brought about the issue and to which one we might revert, if vital.
3. ACID deployment: for each scheduler workflow definition change, we want to have all of the workflow belongings bundled in an atomic, sturdy, remoted and constant method. This manner, if vital, all we have to know is which model of the workflow to roll again to, and the remainder can be taken care of for us.

Whereas all of the above objectives are our North Star, we additionally don’t need to negatively have an effect on quick deployment, excessive availability and arbitrary life span of any deployed asset.

The essential method to flattening arbitrary workflow assets throughout workflow execution has been identified to mankind because the invention of cron, and with the appearance of “infinite” cloud storage programs like S3, this method has served us for a few years. Its obvious flexibility and comfort can usually idiot us into considering that by merely changing the asset within the S3 location we are able to, with none problem, introduce modifications to our enterprise logic. This methodology usually proves very troublesome particularly if there may be a couple of engineer engaged on the identical pipeline and they don’t seem to be all conscious of the opposite of us’ “deployment course of”.

The marginally improved method is proven on the diagram under.

In Determine 1, you possibly can see an illustration of a typical deployment pipeline manually constructed by a person for a person challenge. The continual deployment instrument submits a workflow definition with tips that could belongings in fastened S3 places. These belongings are then individually deployed to those fastened places. At runtime, the belongings are retrieved from the outlined places in S3 and executed within the runtime container. Regardless of requiring customers to assemble the deployment pipeline manually, usually by writing their very own scripts from scratch, this design works and has been efficiently utilized by many groups for years. That being stated, it does have some drawbacks which might be revealed as you attempt to add any quantity of complexity to your deployment logic. Let’s focus on a couple of of them.

Doesn’t contemplate department/PR deployments

In any manufacturing pipeline, you need the pliability of getting a “secure” various deployment logic. For instance, you might need to construct your Scala code and deploy it to another location in S3 whereas pushing a sandbox model of your workflow that factors to this various location. One thing this easy will get very sophisticated in a short time and requires the person to contemplate numerous issues. The place ought to this various location be in S3? Is a single location sufficient? How do you arrange your deployment logic to know when to deploy the workflow to a take a look at or dev setting? Solutions to those questions usually find yourself being extra customized logic inside the person’s deployment scripts.

Can not rollback to earlier workflow variations

Whenever you deploy a workflow, you actually need it to encapsulate an atomic and idempotent unit of labor. A part of the explanation for that’s the want for the power to rollback to a earlier workflow model and realizing that it’ll at all times behave because it did in earlier runs. There might be many causes to rollback however the typical one is whenever you’ve acknowledged a regression in a latest deployment that was not caught throughout testing. Within the present design, reverting to a earlier workflow definition in your scheduling system just isn’t sufficient! It’s important to rebuild your belongings from supply and transfer them to your fastened S3 location that your workflow factors to. To allow atomic rollbacks, you possibly can add extra customized logic to your deployment scripts to at all times deploy your belongings to a brand new location and generate new pointers in your workflows to make use of, however that comes with larger complexity that always simply doesn’t really feel price it. Extra generally, groups will decide to do extra testing to attempt to catch regressions earlier than deploying to manufacturing and can settle for the additional burden of rebuilding all of their workflow dependencies within the occasion of a regression.

Runtime dependency on user-managed cloud storage places

At runtime, the container should attain out to a user-defined storage location to retrieve the belongings required. This causes the user-managed storage system to be a important runtime dependency. If we zoom out to have a look at a whole workflow administration system, the runtime dependencies can change into unwieldy if it depends on varied storage programs which might be arbitrarily outlined by the workflow builders!

Within the try to ship a easy and strong resolution to the managed workflow deployments we created a command line utility known as Dataflow. It’s a Python based mostly CLI + library that may be put in anyplace contained in the Netflix setting. This utility can construct and configure workflow definitions and their belongings throughout testing and deployment. See under diagram:

In Determine 2, we present a variation of the standard manually constructed deployment pipeline. Each asset deployment is launched to some newly calculated UUID. The workflow definition can then establish a particular asset by its UUID. Deploying the workflow to the scheduling system produces a “Deployment Bundle”. The bundle consists of the entire belongings which have been referenced by the workflow definition and your complete bundle is deployed to the scheduling system. At each scheduled runtime, the scheduling system can create an occasion of your workflow with out having to assemble runtime dependencies from exterior programs.

The asset administration system that we’ve created for Dataflow offers a powerful abstraction over this deployment design. Deploying the asset, producing the UUID, and constructing the deployment bundle is all dealt with mechanically by the Dataflow construct logic. The person doesn’t want to pay attention to something that’s occurring on S3, nor that S3 is getting used in any respect! As a substitute, the person is given a versatile UUID referencing system that’s layered on high of our scheduling system’s workflow DSL. Later within the article we’ll cowl this referencing system in some element. However first, let’s take a look at an instance of deploying an asset and a workflow.

Deployment of an asset

Let’s stroll by means of an instance of a workflow asset construct and deployment. Let’s assume we now have a repository known as stranger-data with the next construction:

.
├── dataflow.yaml
├── pyspark-workflow
│ ├── foremost.sch.yaml
│ └── hello_world
│     ├── ...
│     └── setup.py
└── scala-workflow
├── construct.gradle
├── foremost.sch.yaml
└── src
├── foremost
│   └── ...
└── take a look at
└── ...

Let’s now use Dataflow command to see what challenge elements are seen:

stranger-data$ dataflow challenge listingPython Belongings:
-> ./pyspark-workflow/hello_world/setup.py
Abstract: 1 discovered.Gradle Belongings:
-> ./scala-workflow/construct.gradle
Abstract: 1 discovered.Scheduler Workflows:
-> ./scala-workflow/foremost.sch.yaml
-> ./pyspark-workflow/foremost.sch.yaml
Abstract: 2found.

Earlier than deploying the belongings, and particularly if we made any modifications to them, we are able to run unit exams to ensure that we didn’t break something. In a typical Dataflow configuration this guide testing is non-obligatory as a result of Dataflow steady integration exams will try this for us on any pull-request.

stranger-data$ dataflow challenge take a look atTesting Python Belongings:
-> ./pyspark-workflow/hello_world/setup.py... PASSED
Abstract: 1 profitable, 0 failed.Testing Gradle Belongings:
-> ./scala-workflow/construct.gradle... PASSED
Abstract: 1 profitable, 0 failed.Constructing Scheduler Workflows:
-> ./scala-workflow/foremost.sch.yaml... CREATED ./.workflows/scala-workflow.foremost.sch.rendered.yaml
-> ./pyspark-workflow/foremost.sch.yaml... CREATED ./.workflows/pyspark-workflow.foremost.sch.rendered.yaml
Abstract: 2 profitable, 0 failed.Testing Scheduler Workflows:
-> ./scala-workflow/foremost.sch.yaml... PASSED
-> ./pyspark-workflow/foremost.sch.yaml... PASSED
Abstract: 2 profitable, 0 failed.

Discover that the take a look at command we use above not solely executes unit take a look at suites outlined in our Scala and Python sub-projects, nevertheless it additionally renders and statically validates all of the workflow definitions in our repo, however extra on that later…

Assuming all exams handed, let’s now use the Dataflow command to construct and deploy a brand new model of the Scala and Python belongings into the Dataflow asset registry.

stranger-data$ dataflow challenge deployConstructing Python Belongings:
-> ./pyspark-workflow/hello_world/setup.py... CREATED ./pyspark-workflow/hello_world/dist/hello_world-0.0.1-py3.7.egg
Abstract: 1 profitable, 0 failed.Deploying Python Belongings:
-> ./pyspark-workflow/hello_world/setup.py... DEPLOYED AS dataflow.egg.hello_world.person.stranger-data.grasp.39206ee8.3
Abstract: 1 profitable, 0 failed.Constructing Gradle Belongings:
-> ./scala-workflow/construct.gradle... CREATED ./scala-workflow/construct/libs/scala-workflow-all.jar
Abstract: 1 profitable, 0 failed.Deploying Gradle Belongings:
-> ./scala-workflow/construct.gradle... DEPLOYED AS dataflow.jar.scala-workflow.person.stranger-data.grasp.39206ee8.11
Abstract: 1 profitable, 0 failed....

Discover that the above command:

• created a brand new model of the workflow belongings
• assigned the asset a “UUID” (consisting of the “dataflow” string, asset kind, asset namespace, git repo proprietor, git repo identify, git department identify, commit hash and consecutive construct quantity)
• and deployed them to a Dataflow managed S3 location.

We are able to verify the present belongings of any given kind deployed to any given namespace utilizing the next Dataflow command:

stranger-data$ dataflow challenge listing eggs --namespace hello_world --deployedChallenge namespaces with deployed EGGS:hello_world
-> dataflow.egg.hello_world.person.stranger-data.grasp.39206ee8.3
-> dataflow.egg.hello_world.person.stranger-data.grasp.39206ee8.2
-> dataflow.egg.hello_world.person.stranger-data.grasp.39206ee8.1

The above listing might turn out to be useful, for instance if we would have liked to seek out and entry an older model of an asset deployed from a given department and commit hash.

Deployment of a workflow

Now let’s take a look on the construct and deployment of the workflow definition which references the above belongings as a part of its pipeline DAG.

Let’s listing the workflow definitions in our repo once more:

stranger-data$ dataflow challenge listing workflowsScheduler Workflows:
-> ./scala-workflow/foremost.sch.yaml
-> ./pyspark-workflow/foremost.sch.yaml
Abstract: 2 discovered.

And let’s take a look at a part of the content material of one among these workflows:

stranger-data$ cat ./scala-workflow/foremost.sch.yaml
...
dag:
- ddl -> write
- write -> audit
- audit -> publish
jobs:
- ddl: ...
- write:
spark:
script: ${dataflow.jar.scala-workflow}
class: com.netflix.spark.ExampleApp
conf: ...
params: ...
- audit: ...
- publish: ...
...

You possibly can see from the above snippet that the write job needs to entry some model of the JAR from the scala-workflow namespace. A typical workflow definition, written in YAML, doesn’t want any compilation earlier than it’s shipped to the Scheduler API, however Dataflow designates a particular step known as “rendering” to substitute the entire Dataflow variables and construct the ultimate model.

The above expression ${dataflow.jar.scala-workflow} implies that the workflow can be rendered and deployed with the most recent model of the scala-workflow JAR out there on the time of the workflow deployment. It’s attainable that the JAR is constructed as a part of the identical repository by which case the brand new construct of the JAR and a brand new model of the workflow could also be coming from the identical deployment. However the JAR could also be constructed as a part of a totally totally different challenge and in that case the testing and deployment of the brand new workflow model might be utterly decoupled.

We confirmed above how one would request the most recent asset model out there throughout deployment, however with Dataflow asset administration we are able to distinguish two extra asset entry patterns. An apparent subsequent one is to specify it by all its attributes: asset kind, asset namespace, git repo proprietor, git repo identify, git department identify, commit hash and consecutive construct quantity. There may be yet another further methodology for a center floor resolution to choose a particular construct for a given namespace and git department, which can assist throughout testing and improvement. All of that is a part of the user-interface for figuring out how the deployment bundle can be created. See under diagram for a visible illustration.

In brief, utilizing the above variables offers the person full flexibility and permits them to choose any model of any asset in any workflow.

An instance of the workflow deployment with the rendering step is proven under:

stranger-data$ dataflow challenge deploy...Constructing Scheduler Workflows:
-> ./scala-workflow/foremost.sch.yaml... CREATED ./.workflows/scala-workflow.foremost.sch.rendered.yaml
-> ./pyspark-workflow/foremost.sch.yaml... CREATED ./.workflows/pyspark-workflow.foremost.sch.rendered.yaml
Abstract: 2 profitable, 0 failed.Deploying Scheduler Workflows:
-> ./scala-workflow/foremost.sch.yaml… DEPLOYED AS https://hawkins.com/scheduler/sandbox:person.stranger-data.scala-workflow
-> ./pyspark-workflow/foremost.sch.yaml… DEPLOYED AS https://hawkins.com/scheduler/sandbox:person.stranger-data.pyspark-workflow
Abstract: 2 profitable, 0 failed.

And right here you possibly can see what the workflow definition appears like earlier than it’s despatched to the Scheduler API and registered as the most recent model. Discover the worth of the script variable of the write job. Within the unique code says ${dataflow.jar.scala-workflow} and within the rendered model it’s translated to a particular file pointer:

stranger-data$ cat ./scala-workflow/foremost.sch.yaml
...
dag:
- ddl -> write
- write -> audit
- audit -> publish
jobs:
- ddl: ...
- write:
spark:
script: s3://dataflow/jars/scala-workflow/person/stranger-data/grasp/39206ee8/1.jar
class: com.netflix.spark.ExampleApp
conf: ...
params: ...
- audit: ...
- publish: ...
...

The Infrastructure DSE group at Netflix is chargeable for offering insights into knowledge that may assist the Netflix platform and repair scale in a safe and efficient approach. Our group members accomplice with enterprise items like Platform, OpenConnect, InfoSec and interact in enterprise degree initiatives regularly.

One aspect impact of such huge engagement is that over time our repository advanced right into a mono-repo with every module requiring a personalized construct, testing and deployment technique packaged right into a single Jenkins job. This setup required fixed repairs and in addition meant each time we had a construct failure a number of folks wanted to spend so much of time in communication to make sure they didn’t step on one another.

Final quarter we determined to separate the mono-repo into separate modules and undertake Dataflow as our asset orchestration instrument. Put up deployment, the group depends on Dataflow for automated execution of unit exams, administration and deployment of workflow associated belongings.

By the top of the migration course of our Jenkins configuration went from:

to:

cd /dataflow_workspace
dataflow challenge deploy

The simplicity of deployment enabled the group to concentrate on the issues they got down to clear up whereas the department based mostly customization gave us the pliability to be our simplest at fixing them.

This new methodology out there for Netflix knowledge engineers makes workflow administration simpler, extra clear and extra dependable. And whereas it stays pretty straightforward and secure to construct your small business logic code (in Scala, Python, and so forth) in the identical repository because the workflow definition that invokes it, the brand new Dataflow versioned asset registry makes it simpler but to construct that code utterly independently after which reference it safely inside knowledge pipelines in some other Netflix repository, thus enabling straightforward code sharing and reuse.

Yet another side of knowledge workflow improvement that will get enabled by this performance is what we name branch-driven deployment. This method permits a number of variations of your small business logic and workflows to be working on the similar time within the scheduler ecosystem, and makes it straightforward, not just for particular person customers to run remoted variations of the code throughout improvement, but additionally to outline remoted staging environments by means of which the code can cross earlier than it reaches the manufacturing stage. Clearly, to ensure that the workflows to be safely utilized in that configuration they need to adjust to a couple of easy guidelines almost about the parametrization of their inputs and outputs, however let’s depart this topic for one more weblog submit.

Particular due to Peter Volpe, Harrington Joseph and Daniel Watson for the preliminary design overview.