Data Engineering��Building Pipelines / Plumbing�Operationalization / DataOps

A Very Pragmatic Topic

In the “real world” you don’t deal with data sets, you deal with data feeds

• Data “arrives” on a regular basis

When data arrives, you need to “do the right thing”

​

Things to consider

• How does data “arrive”?
• What kinds of things do you need do upon arrival?
• How do you automate the things?
• How do you maintain quality and handle exceptions?

​

​

How does data arrive?

Push:

• Files appear in a staging folder, e.g. in AWS S3
• Tables get populated in a data warehouse, e.g. in BigQuery
• A software system API provides the data
• E.g. publish-subscribe or queueing systems, as we’ll discuss

​

• Pull: Extract data from some source
• Via scheduling software
• On receipt of an event, e.g. from pub-sub or queueing systems
• Or slack or email or … really anything

What do you do upon Data Arrival?

• Run data quality (DQ) tests
• E.g. check the new data has expected schema, integrity constraints, data cleaning checks
• Load into “the right place” (a data warehouse or a curated set of directories)
• Run transformation logic
• Log Metadata (Data Lineage): track what arrived from whom, where, when…
• Pass along info/control to humans

Isn’t this all Data Transformation?

• Yes!! Data Engineering is largely automated data transformation
• But in many shops this involves tasks across many software systems
• Our SQL-centric ELT this semester has been very clean
• Not unrealistic -- some shops do this for their main operations
• Many pipelines involve different systems for some/all pieces of the puzzle
• Derived data: Preparation/Cleaning/Transformation systems, Entity Resolution/Standardization systems
• Data Structure Maintenance: Indexes, Materialized View updates, Data Cube updates
• Data Security/Compliance Systems: “masking” private data, encryption
• Metadata storage: catalogs, data lineage (tracking where data came from and what happened)
• Etc. etc.
• Need to automate these many tasks/systems and their handoffs
• And deal with exceptions, human interventions/approvals, etc.

How do we automate the work?

• More kinds of systems! (Ugh)
• Workflow Systems: manage processes (human or software)
• With process dependencies (order of activity)
• E.g. Activiti, Airflow, BPL, Luigi, …
• Dataflow Systems: manage data movement through operations (software)
• With data dependencies (inputs/outputs)
• E.g. Spark, dbt + SQL, Flink, Dataflow
• Messaging/Queueing/Eventing: hand off data reliably from one system to another
• E.g. Kafka, RabbitMQ, AWS SQS, MSMQ, MQSeries, …

​

• Many times you will use combinations of these
• And it can be a matter of taste/experience which to use where

What about quality and exceptions

• Data Quality is both similar to and different from Software Quality

​

• The notion of a “test” is data-centric: logical (schemas, constraints, etc), statistical (outliers, distributional drift) etc.
• Quite different from software debugging

​

• But SW Eng DevOps principles are getting more play in DataOps pipelines
• Test-driven development
• Continuous Integration / Continuous Deployment
• We’ll look at dbt as an emerging example

Agenda

• A quick intro to Pub-Sub and MQ systems
• A quick intro to Workflow systems
• A demo of SQL-based Dataflow and testing using dbt
• Relevant to final project (MS or extra credit)

Two Slides on Pub-Sub / Message Queues

• Need: Handoff data between producer and consumer processes in a pipeline
• Asynchronously
• A data storage problem: asynchrony requires persistent storage to buffer whichever comes first
• Producer or Consumer
• Reliably
• Producer wants to make sure message has been queued durably
• Consumer wants to get every message, and handle it to completion atomically
• Ideally: Exactly-Once delivery
• Atomic Enqueue, Atomic Dequeue-and-process
• Queue acknowledges durable receipt from producer
• Client acknowledges commit of a message-handling transaction
• Can be done with “distributed” transactions across processes and queue
• In many cases: At-Least-Once delivery
• Client takes responsibility to ensure idempotent message handling

Current Landscape

• Publish-Subscribe: (a.k.a. message bus) typified by Kafka
• Persistent subscriptions to topics like “Updates-to-table-X”
• A stream of messages labeled with one or more topics
• A durable log (as big as many databases) holding a window of past events
• E.g. in Kafka can request a batch of messages from some offset in past
• Optimized for high-throughput (lots of data)

​

• Persistent Message Queue typified by RabbitMQ
• Register “routes” for forwarding messages to consumers
• Stream of messages to be routed
• No big log – each message deleted after (acknowledged) handoff
• Optimized for low-latency (speed per message)

Can I just use a transactional database for this?

It’s doable in SQL, but DBMSs are not tuned for this workload.

Two Slides on Workflow Systems

• Need: orchestrate a bunch of separate processes or API calls
• Control order of invocation
• Programming constructs like conditionals (IF) and loops over system
• Handle (potentially long) asynchrony in request/response
• Allow for steps that involve humans
• Run reliably over long periods of time, despite failures of components or the workflow machines
• Often storing the state of the workflow in a SQL DBMS
• Sometimes guaranteeing transition between steps transactionally
• Provide features to schedule and monitor progress

Example: Apache Airflow

• A DAG of (ideally idempotent) tasks passing small amounts of data
• DAGs are constructed programmatically in Python
• Simple syntax for defining tasks and setting up “edges” between them
• DAG “shape” can be dynamic at runtime
• Exception logic to fail, skip, reschedule a task
• Tasks (Nodes) in the DAG are user-defined
• Python code, bash calls, �emails, Slack msgs, API calls, etc.
• Scales up
• Via use of message queues
• Schedules and tracks Runs
• “Sensors” to check for time passing�or data arriving in filesystem or database

​

Dataflow: Spark, Google Dataflow, etc.

• Like Pandas, lets you define dataflows functionally

​

​

​

​

​

• Spark “RDDs” are akin to Database Views
• A record of the code needed to compute results from data on demand
• But in a dataflow format akin to a query plan in a database
• I.e.EXPLAIN in Postgres

​

Advantage of Algebras over SQL for Transformation

• Writing code in algebraic forms (Rel Alg, Spark, Pandas) is incremental
• Build up your complex transformation pipelines one step at a time
• Good for exploratory thinking, debugging, incremental complexity
• SQL wants us to “declare” our outputs
• Discourages stepwise reasoning?
• CTEs/WITH and views can help
• But syntax is unwieldy
• CTEs are inconvenient to debug
• Still, these distinctions are form, not function
• SQL can do about anything that algebras can do
• Standard, clean semantics, portable (even to Spark these days!)
• Transactional support, lots of tools, etc.
• What can we do about SQL’s “ergonomics” for writing complex Transformation scripts

Data Pipelines in SQL: dbt

• dbt (the “data build tool”) is a commercial open-source project
• free to use the CLI (command-line interface) version
• pay Fishtown for cloud-based scheduling, CI-git integration, GUI
• various other companies standardizing on dbt format, including Trifacta

​

• a declarative specification framework
• for pipelines of ELT-style SQL transformations
• for data quality tests
• specified in YAML

​

• We’ll do a quick tour today!

​

A Basic dbt Repo 1/4

• Expectation is that you connect this to an existing�Data Warehouse (i.e. SQL database)

A Basic dbt Repo 2/4

• Models
• SQL Transformations
• 1 query per file
• schema.yml
• optional metadata and integrity tests
• dbt_project.yml
• required
• standard boilerplate generated for you
• set your defaults for config options

A Basic dbt Repo 3/4

• Tests
• SQL query per file
• Passes if it returns 0 rows!
• Data: Csv’s for “seed data” (e.g. static lookup tables)
• Analysis: Extra SQL queries you don’t need to run
• Macros: String-based macros in the Jinja language
• Snapshots: SQL queries to materialize historical data� as it’s being updated

​

A Basic dbt Repo 4/4

• Also ~/.dbt/profiles.yml
• DB connectivity information
• Per dbt “profile”
• Matches your dbt_project.yml

A Simple dbt project for Project 2

• Lines 5-7 configure handy metadata
• Line 10 specifies the connection in ~/.dbt/profiles.yml
• Lines 12-25 are boilerplate for the standard directory hierarchy generated by dbt init
• Lines 34-39 provide defaults for our models
• contained in a database named “baseball”
• by default, model results are views, and not materialized

A Simple Model

• Specifies a derived result named q1a
• By default, this will be materialized as a view

Source References

• Specifies a derived result named q1b
• By default, this will be materialized as a view
• Note the FROM clause:
• Expressions in squiggly brackets are macros in the Jinja language
• Source is a built-in macro that looks up a table name in yaml files under the models directory
• Allows a level of indirection so you can retarget your models to another database or schema or whatever!
• Nice hygiene

Source References

Model Materialization:�Table

• When we run this project, inducted_hof_ca_mat becomes a table rather than a view
• Recomputed every time we run the project!

Model Materialization:�Incremental

• q3a_innerjoin becomes a table on first run
• On subsequent runs, lines 10 & 11 are added to limit the work
• A simplistic “materialized view maintenance” facility outside the database

Model Materialization:�Incremental

• q3a_innerjoin becomes a table on first run
• On subsequent runs, lines 10 & 11 are added to limit the work
• A simple “materialized view maintenance” facility outside the database
• Only handles inserts and “upserts” (replace tuples by a unique key)
• No delete handling, no general update handling
• Up to you to figure out if it’s a performance win
• Principle from early lectures: “selection pushdown is good”
• But it depends on the DBMS you’re using (see the docs)

ref and pipelines

• One model can reference another
• {{ ref(‘<model name>’) }}
• Implicitly specifies a pipeline
• Models querying models
• Cannot have cycles of reference!
• Must be a DAG

Lineage DAG in dbt docs

• dbt docs generate
• dbt docs serve
• Note that there’s no notion of “tasks” here, just data lineage (data “dependencies”)
• We dbt run the whole DAG
• All non-incremental models are rebuilt from scratch each time
• Incremental models capture the notion of “change propagation”
• Though fairly primitive in dbt
• There are custom products that are better at “Change Data Capture” and materialized view maintenance
• Other nice stuff in dbt docs, poke around.

Jinja Macros

• Recall that SQL only lets you query rows
• SELECT and FROM lists are “hard-wired”, not dynamic
• Jinja can help
• Essentially, parameterized queries via string macros

​

• A blunt instrument
• Easy to write queries that don’t work or are undesirable
• Keep it simple
• Can pull in libraries like dbt_utils from git
• packages.yml at top level
• E.g. star command in the upper�right lets you choose which�columns you want in the �SELECT list more flexibly

​

Running Your Models

Tests

• Schema tests in .yml files

​

• General “data” tests in .sql files

Schema Tests

• Under the models key in the .yml files
• Like SQL DDL constraints, but mid-pipeline
• unique, not_null, accepted_values and relationships
• single-column by default but kludges available to test composite columns

“Data” Tests

• Just plain old .sql files
• any check you like! but keep it quickish…
• return no rows when test passes
• return something small otherwise

Running Tests

How to Debug Models & Tests

Look in:

• target/compiled
• models and tests after Jinja processing complete
• run
• the DDL that runs to build the models

​

• Can clean this all up with dbt clean

Transactional Concerns in dbt

• By default, each model execution is a transaction
• But the DBMS computes views inside whatever query uses them
• So materialization via view vs table/incremental affects transaction boundaries
• Can specify pre-/post-“hooks” on models to run more SQL statements before/after
• And can specify each hook as outside or inside the model’s transaction

DataOps: CI Testing

• Continuous Integration

​

​

• Make sure every code checkin forces dbt test to run
• Force a test at client prior to github commit with commit hooks
• You can pay for Fishtown CI integration of dbt with github
• Or you can use a github-integrated CI tool like Travis/Jenkins
• Pretty easy to get it to fire off dbt test
• Need to ensure database connectivity in all cases

Dev/Test/Prod

• Standard CI/CD has code go through 2 or 3 different deployments
• Development
• Test or CI
• Production
• In a data warehouse, typically do this with different schemas or databases for each
• Specified via outputs and target in the ~/.dbt/profiles.yml file
• Queries sources and generates outputs in the appropriate schema/database
• Need to keep track of which version of the code (git tag) is associated with each deployment