As part of Unified Analytics Platform, Databricks Workspace along with Databricks File System (DBFS) are critical components that facilitate collaboration among data scientists and data engineers: Databricks Workspace manages users’ notebooks, whereas DBFS manages files; both have REST API endpoints to manage notebooks and files respectively.

While the REST APIs are principally designed for general programmatic use, ad-hoc tasks of exploring and manipulating desired content is not that easy with REST APIs without writing repetitive boilerplate code. So to simplify this task for Databricks developers, we have implemented an easy command line interface that interacts with Databricks workspaces and filesystem APIs.

In part 1 of our blog aimed at Databricks developers, we outline some use cases where you can employ a command line interface to interact with Databricks workspaces and manipulate files and folders.

Usage Scenarios

Checking in notebooks to version control systems (VCS)

One of the most common usages of the Databricks CLI is to enable an alternative integration point to VCS.

To check notebooks into VCS, you need a canonical text-based file for the notebook. With the workspace APIs, you can export and import notebooks to this canonical text-based file format. The Databricks CLI builds on this idea further by wrapping these APIs into an easy to use command line interface with support for recursive import and export.

For example, consider a scenario with two users’ workspace and a production workspace: Alice with workspace A, Bob with workspace B, and a production workspace P with notebooks that are run through Databricks Job Scheduler. After developing code in her workspace A, Alice may export her code with databricks workspace export_dir to her git repository and initiate a pull request. Bob can then review and approve the PR, after which Alice can merge her changes to the master. This merge will trigger a Continuous Delivery job in which production workspace P will initiate a databricks workspace import_dir, bringing all new changes into production.

Copying small datasets to DBFS

Another use case for the CLI is importing small datasets to DBFS.

For example:

• (Recursively) copying datasets/files between local file system and DBFS

Installation

To begin, install the CLI by running the following command on your local machine.
pip install --upgrade databricks-cli

Note that the Databricks CLI currently cannot run with Python 3.

After installation is complete, the next step is to provide authentication information to the CLI. The first and recommended way is to use an access token generated from Databricks. To do this run databricks configure --token. A second way is to use your username and password pair. To do this run databricks configure and follow the prompts. After following the prompts, your access credentials will be stored in the file ~/.databrickscfg.

Workspace CLI examples

The implemented commands for the Workspace CLI can be listed by running databricks workspace -h. Commands are run by appending them to databricks workspace. To make it easier to use the workspace CLI, feel free to alias databricks workspace to something shorter. For more information reference Aliasing Command Groups and Workspace API.

$ databricks workspace -h Usage: databricks workspace [OPTIONS] COMMAND [ARGS]... Utility to interact with the Databricks Workspace. Workspace paths must be absolute and be prefixed with `/`. Options: -v, --version -h, --help Show this message and exit. Commands: delete Deletes objects from the Databricks... export Exports a file from the Databricks workspace... export_dir Recursively exports a directory from the... import Imports a file from local to the Databricks... import_dir Recursively imports a directory from local to... list List objects in the Databricks Workspace ls List objects in the Databricks Workspace mkdirs Make directories in the Databricks Workspace. rm Deletes objects from the Databricks...

Listing Workspace Files

Exporting a workspace directory to the local filesystem

The databricks workspace export_dir command will recursively export a directory from the Databricks workspace to the local filesystem. Only notebooks are exported and when exported, the notebooks will have the appropriate extensions (.scala, .py, .sql, .R) appended to their names).

Importing a local directory of notebooks

Similarly, the databricks workspace import_dir command will recursively import a directory from the local filesystem to the Databricks workspace. Only directories and files with the extensions of .scala, .py, .sql, .r, .R are imported. When imported, these extensions will be stripped off the name of the notebook.

To overwrite existing notebooks at the target path, the flag -o must be added.

DBFS CLI Examples

The implemented commands for the DBFS CLI can be listed by running databricks fs -h. Commands are run by appending them to databricks fs and all dbfs paths should be prefixed with dbfs:/. To make the command less verbose, we’ve gone ahead and aliased dbfs to databricks fs. For more information reference DBFS API.

$ databricks fs -h Usage: databricks fs [OPTIONS] COMMAND [ARGS]... Utility to interact with DBFS. DBFS paths are all prefixed with dbfs:/. Local paths can be absolute or local. Options: -v, --version -h, --help Show this message and exit. Commands: configure cp Copy files to and from DBFS. ls List files in DBFS. mkdirs Make directories in DBFS. mv Moves a file between two DBFS paths. rm Remove files from dbfs.

Copying a file to DBFS

It’s possible to copy files from your localhost to DBFS both file by file and recursively. For example to copy a CSV to DBFS, you can run the following command.

For recursive copying, add the -r flag.

Copying a file from DBFS

Similarly, it is possible to copy files from DBFS back to the local filesystem.

What’s Next

We’re actively developing new features for the Databricks CLI for developers. The next item immediately on our roadmap is to support cluster and jobs APIs endpoints. We will cover in the second part of this blog series.

In the spirit of our open source Apache Spark heritage, the source code for the CLI is released on Github. If you have feedback on a feature, please leave an issue on our Github project.

Try Databricks for free. Get started today

Originally published at databricks.com on November 8, 2017.

Introducing Command Line Interface for Databricks Developers was originally published in Databricks Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Original Authors: Hossein Falaki and Xiangrui Meng

At Databricks we strive to make our Unified Analytics Platform the best place to run big data analytics. For big data, Apache Spark has become the de-facto computing engine, while for advanced analytics, R is one of the most widely used languages and environments. R’s package ecosystem boasts more than 10k packages ranging from implementation of simple statistical functions to sophisticated packages in verticals such as genomics and finance.

The Unified Analytics Platform, with Databricks Runtime (DBR) at its core, accelerates and unifies the strengths of both Apache Spark and R. The DBR helps customers, in a wide range of industries and verticals, to extract value from their big data efficiently.

Many Databricks R users take advantage of the strengths of Apache Spark and R’s rich ecosystem with a two-step workflow. First, they perform all large data operations on distributed SparkDataFrames using SparkR API. These usually include loading data from its sources, parsing and transforming it into desired format and shape. In many cases, the interesting parts of the final structured dataset can fit in a single machine’s memory. At this stage, users convert distributed datasets to a local R data.frames and pass them to other (single-node) R packages for further analysis or visualization. Often, the conversions between R data.frames and Spark DataFrames happen multiple times. For example, results of an R package functions are parallelized and joined with distributed datasets.

In this blog post, we introduce two new improvements in Databricks Runtime 3.3 (DBR) that accelerate these common workflows. First, we added support for R packages as part of Databricks library management. Second, as part of our DBIO accelerator module, we have accelerated the performance of SparkR::collect() and SparkR::createDataFrame(). These two APIs are the bridges between single-node R and distributed Spark applications and are among the most frequently used functions of SparkR.

R Package Management in Databricks

On Databricks workspace, users can now define a library that points to their desired CRAN repository and package. When this library is attached to a cluster, all workers and the driver node will install the CRAN package automatically. This functionality is accessible through the REST API as well.

With managed R libraries, integrating workflows with third-party R packages will be much easier on Databricks. This is especially the case with auto-scaling clusters where new workers may be added to a cluster dynamically.

High-performance SparkR with DBIO

We used the airlines dataset for the benchmark. The dataset consists of over 120 million rows and 29 columns of numeric and character types in CSV format, which is common and popular among R users. We progressively used larger fractions of the dataset to evaluate throughput and latency with varying data size and also found the limit after which the calls fail. We compared DBR 3.3 with DBR 3.0 on clusters comprising of four i3.xlarge workers.

Measuring SparkR::collect() performance

We first load data using Spark’s CSV data source with automatic schema inference. We cache and materialize the Spark DataFrame and then collect it to a local R data.frame. We measure the total elapsed time of the collect step.

Measuring SparkR::createDataFrame() performance

We load the file from the local file system using R’s tabular data reading functionality with a default configuration, which automatically infers schema. We then parallelize the data.frame with SparkR::createDataFrame() and count the number of rows. For this benchmark, We measure elapsed time of last two steps combined.

Benchmark Results

First, we compare average throughput of parallelizing R data.frames on DBR 3.3 and DBR 3.0. DBIO can achieve 300x higher average throughput in DBR 3.3 compared to DBR 3.0. When collecting Spark DataFrames, we observed 24X higher average throughput on Databricks Runtime 3.3 compared to the older version.

The plots below show end-to-end latency. This is what users perceive when calling SparkR API. On each DBR version, we progressively increased input size until the call would fail. We measured elapsed time for each successful run.

On DBR 3.0 SparkR::createDataFrame() failed with data larger than 750K rows (about 70MB). On DBR 3.3 the call did not fail for any R data.frame. Overall createDataFrame() is 100X faster on DBR 3.3— and can handle much larger data.

The failure point of SparkR::collect() has not changed. At about 6M rows (550MB) the R process cannot handle the single in-memory object, and we observed failure when collecting Spark DataFrames. In this experiment, DBR 3.3 is 10x faster than older versions across varying input size.

Conclusions

We are continuously working to improve different R workflows on Databricks. We recently announced integration with sparklyr; R package management and improved SparkR performance are our most recent steps toward that goal.

As shown above, the DBIO in Databricks Runtime 3.3 significantly accelerates the performance of two of the most important SparkR calls: SparkR::collect() and SparkR::createDataFrame(). These calls transfer data from Spark’s JVM to R and vice-versa and are the most popular SparkR APIs.

Read More

To read more about our efforts with SparkR on Databricks, we refer you to the following assets:

Try Databricks for free. Get started today

Originally published at databricks.com on October 6, 2017.

Accelerating R Workflows on Databricks was originally published in Databricks Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Original authors: Ron Hu, Zhenhua Wang, Wenchen Fan, and Sameer Agarwal

This is a joint engineering effort between Databricks’ Apache Spark engineering team (Sameer Agarwal and Wenchen Fan) and Huawei’s engineering team (Ron Hu and Zhenhua Wang)

Apache Spark 2.2 recently shipped with a state-of-art cost-based optimization framework that collects and leverages a variety of per-column data statistics (e.g., cardinality, number of distinct values, NULL values, max/min, average/max length, etc.) to improve the quality of query execution plans. Leveraging these statistics helps Spark to make better decisions in picking the most optimal query plan. Examples of these optimizations include selecting the correct build side in a hash-join, choosing the right join type (broadcast hash-join vs. shuffled hash-join) or adjusting a multi-way join order, among others.

In this blog, we’ll take a deep dive into Spark’s Cost Based Optimizer (CBO) and discuss how Spark collects and stores these statistics, optimizes queries, and show its performance impact on TPC-DS benchmark queries.

A Motivating Example

At its core, Spark’s Catalyst optimizer is a general library for representing query plans as trees and sequentially applying a number of optimization rules to manipulate them. A majority of these optimization rules are based on heuristics, i.e., they only account for a query’s structure and ignore the properties of the data being processed, which severely limits their applicability. Let us demonstrate this with a simple example. Consider a query shown below that filters a table t1 of size 500GB and joins the output with another table t2 of size 20GB. Spark implements this query using a hash join by choosing the smaller join relation as the build side (to build a hash table) and the larger relation as the probe side 1. Given that t2 is smaller than t1, Apache Spark 2.1 would choose the right side as the build side without factoring in the effect of the filter operator (which in this case filters out the majority of t1‘s records). Choosing the incorrect side as the build side often forces the system to give up on a fast hash join and turn to sort-merge join due to memory constraints.

Apache Spark 2.2, on the other hand, collects statistics for each operator and figures out that the left side would be only 100MB (1 million records) after filtering, and the right side would be 20GB (100 million records). With the correct size/cardinality information for both sides, Spark 2.2 would choose the left side as the build side resulting in significant query speedups.

In order to improve the quality of query execution plans, we enhanced the Spark SQL optimizer with detailed statistics information. From the detailed statistics information, we propagate the statistics to other operators as we traverse the query tree upwards. Once propagated, we can estimate the number of output records and output size for each database operator, enabling and optimizing an efficient query plan.

Statistics Collection Framework

ANALYZE TABLE command

CBO relies on detailed statistics to optimize a query plan. To collect these statistics, users can issue these new SQL commands described below:

ANALYZE TABLE table_name COMPUTE STATISTICS

The above SQL statement can collect table level statistics such as number of rows and table size in bytes. Note that ANALYZE, COMPUTE, and STATISTICS are reserved keywords and can take specific column names as arguments, storing all the table level statistics in the metastore.

ANALYZE TABLE table_name COMPUTE STATISTICS FOR COLUMNS column-name1, column-name2, ….

Note that it’s not necessary to specify every column of a table in the ANALYZE statement — only those that are used in a filter/join condition, or in group by clauses etc.

Types of Statistics

The following tables list the detailed statistics information we collect for numeric/date/timestamp and string/binary data types respectively.

Given that CBO traverses Spark’s LogicalPlan in a post-order fashion, we can propagate these statistics bottom-up to other operators. Although there are numerous operators for which we can estimate their statistics and corresponding costs, here we’ll describe the process to estimate statistics for two of the most complex and interesting operators: FILTER and JOIN.

FILTER Selectivity

A filter condition is a predicate expression specified in the WHERE clause of a SQL SELECT statement. A predicate can be a compound logical expression with logical AND, OR, NOT operators combining multiple single conditions. A single condition usually has comparison operators such as =, <, <=, >, >= or <=>2. Therefore, it can be quite complex to estimate the selectivity for the overall filter expression.

Let’s examine a few computations of filter selectivity for a compound logical expression comprising of multiple single logical expressions.

• For logical AND expression, its filter selectivity is the selectivity of left condition multiplied by the selectivity of the right condition, i.e., fs(a AND b) = fs(a) * fs (b).
• For logical-OR expression, its filter selectivity is the selectivity of left condition, plus the selectivity of the right condition, and minus the selectivity of left condition logical-AND right condition,
  i.e., fs (a OR b) = fs (a) + fs (b) - fs (a AND b) = fs (a) + fs (b) – (fs (a) * fs (b))
• For logical-NOT expression, its filter factor is 1.0 minus the selectivity of the original expression, i.e., fs (NOT a) = 1.0 - fs (a)

Next, let’s examine a single logical condition that can have comparison operators, such as =, <, <=, >, >= or <=>. In order to drive intuition, we will compute the filter selectivity for equality (=) and less than (<) operators for the case with one operand as column and the other operand as a literal. Other comparison operators can be derived similarly.

• Equal-to (=) condition: Here we check if the literal constant value of the condition falls within the qualified range between the current minimal and maximal column value. This is necessary because the range may change due to a previous condition applied earlier. If the constant value is outside the qualified range, then the filter selectivity is simply 0.0. Otherwise, it is the inverse of the number of distinct values (note that without additional histogram information, we just assume a uniform distribution for the column values). Future releases will leverage histograms to further improve estimation accuracy.
• Less-than (<) condition: Here we check where the constant literal value of the condition falls. If it is smaller than the current minimal column value, then the filter selectivity is simply 0.0 (and if it’s larger than the current maximal value, then the selectivity is just 1.0). Otherwise, we compute the filter factor based on the available information. If there is no histogram, we prorate and set filter selectivity to: (constant – minimum) / (maximum – minimum). On the other hand, if there is a histogram, we can calculate the filter selectivity by adding up the density of histogram buckets between current column minimal values and the constant value. Also, note that the constant on the right hand side of the condition would now become the new maximal column value.

Join Cardinality

Now that we’ve discussed filter selectivity, let’s talk about join output cardinality. Before computing the cardinality of a two-way join output, we should already have the output cardinalities of both its child nodes. The cardinality of each join side is no greater than the number of records in the original join table; rather, it is the number of qualified records after applying all execution operators before this join operator. Here we are specifically interested in computing the cardinality of inner join operation as it is often used to derive the cardinalities of other join types. We compute the number of rows of A join B on A.k = B.k as:

num(A IJ B) = num(A)*num(B)/max(distinct(A.k),distinct(B.k))

where num(A) is the number of qualified records in table A immediately before join operation, and distinct is the number of distinct values of the join column k.

By computing the inner join cardinality, we can similarly derive the join cardinality for other join types as below:

• Left-Outer Join: num(A LOJ B) = max(num(A IJ B),num(A)) meaning the larger value between the inner join output cardinality and cardinality of the left join outer side A. This is because we still need to factor in every record of the outer side although some of them do not produce join output records.
• Right-Outer Join: num(A ROJ B) = max(num(A IJ B),num(B))
• Full-Outer Join: num(A FOJ B) = num(A LOJ B) + num(A ROJ B) - num(A IJ B)

Optimal Plan Selection

Now that we are armed with intermediate data statistics, let us discuss how Spark SQL uses this information to pick the best possible query plan. Earlier we explained selecting build side based on accurate cardinality and statistics for a hash join operation.

Similarly, with accurate cardinality and size estimation for all operators preceding a join operator, we can better estimate the size of a join side to decide whether or not that side can meet the broadcast criteria.

These statistics also help us leverage cost-based join reordering optimizations. We adapted the dynamic programming algorithm in [Selinger 1979]3to pick the best m-way join order. To be more specific, when building m-way joins, we only keep the best plan (with the lowest cost) for the same set of m items. For instance, for 3-way joins, we only keep the best join plan with its order for items {A, B, C} among plans (A join B) join C, (A join C) join B and (B join C) join A. Our adapted algorithm considers all combinations including left-deep trees, bushy trees, and right-deep-trees. We also prune cartesian product candidates when building a new plan if there exists no join condition involving references from both left and right subtrees. This pruning strategy significantly reduces the search space.

Most database optimizers factor in the CPU and I/O costs separately to estimate the overall operator cost. In Spark, we estimate the cost of a join operator with a simple formula:

cost = weight * cardinality + (1.0 - weight) * size 4

The first portion of the join cost formula roughly corresponds to the CPU cost and the second portion roughly corresponds to the I/O cost. The cost of a join tree is the summation of costs of all intermediate joins.

Query Benchmark and Analysis

We took a non-intrusive approach while adding these cost-based optimizations to Spark by adding a global config spark.sql.cbo.enabled to enable/disable this feature. In Spark 2.2, this parameter is set to false by default. This was a conscious short-term choice keeping in mind that Spark is being used in production at thousands of companies and changing the default behavior might be undesirable for existing production workloads.

Configuration and Methodology

We ran all TPC-DS benchmark queries with Apache Spark 2.2 on a 4 node cluster (Huawei FusionServer RH2288 with 40 cores and 384 GB memory each) and a scale factor of 1000 (i.e., 1TB data). It took us ~14 minutes to collect the statistics on all 24 tables (and a total of 425 columns).

Before examining the end-to-end results, let us first look at a specific TPC-DS query (Q25; shown below) to better understand the impact of cost-based join ordering. This query involves three fact tables: store_sales (2.9 billion rows), store_returns (288 million rows) and catalog_sales (1.44 billion rows). It also involves three dimension tables: date_dim (73K rows), store (1K rows) and item (300K rows).

SELECT i_item_id, 
 i_item_desc, 
 s_store_id, 
 s_store_name, 
 sum(ss_net_profit) AS store_sales_profit, 
 sum(sr_net_loss) AS store_returns_loss, 
 sum(cs_net_profit) AS catalog_sales_profit 
FROM 
 store_sales, store_returns, catalog_sales, date_dim d1, date_dim   d2, date_dim d3, store, item 
WHERE 
 d1.d_moy = 4
   AND d1.d_year = 2001
   AND d1.d_date_sk = ss_sold_date_sk
   AND i_item_sk = ss_item_sk
   AND s_store_sk = ss_store_sk
   AND ss_customer_sk = sr_customer_sk
   AND ss_item_sk = sr_item_sk
   AND ss_ticket_number = sr_ticket_number
   AND sr_returned_date_sk = d2.d_date_sk
   AND d2.d_moy BETWEEN 4 AND 10
   AND d2.d_year = 2001
   AND sr_customer_sk = cs_bill_customer_sk
   AND sr_item_sk = cs_item_sk
   AND cs_sold_date_sk = d3.d_date_sk
   AND d3.d_moy BETWEEN 4 AND 10
   AND d3.d_year = 2001
GROUP BY
 i_item_id, i_item_desc, s_store_id, s_store_name
ORDER BY
 i_item_id, i_item_desc, s_store_id, s_store_name
LIMIT 100

Q25 without CBO

Let us first look at the unoptimized join tree for Q25 without CBO (shown below). This is also commonly referred to as a left-deep tree. Here, joins #1 and #2 are large fact-to-fact table joins that join the 3 fact tables store_sales, store_returns, and catalog_sales together, and produce very large intermediate tables. Both these joins executed as shuffle joins with the large join #1 outputting 199 million rows. Overall, the query takes 241 seconds when CBO is disabled.

Q25 with CBO

On the other hand, with cost-based optimization, Spark creates an optimal join plan that reduces intermediary data size (shown below). In this case, Spark creates a bushy tree instead of a left-deep tree. When CBO is enabled, Spark joins the fact tables with their corresponding date_dim dimension table first (before attempting any fact-to-fact joins). Avoiding large fact-to-fact joins means avoiding large expensive shuffles. In this query, this reduces the intermediate data size to about 1/6th (compared to the previous case). As a result, Q25 took only 71 seconds, representing a speedup of 3.4X.

TPC-DS Query Performance

Now that we’ve established an intuition behind where the improvements come from, let us look at the end-to-end TPC-DS query results. The figure below shows query times for all TPC-DS queries with and without CBO:

First of all, note that about half of TPC-DS benchmark queries do not show performance change. This is because there is no change in the query plan when running queries with or without CBO (i.e., even without CBO, existing heuristics in Spark’s Catalyst optimizer were able to optimize these queries well). For the remaining queries, while there were improvements across the board, what’s perhaps more interesting is the set of 16 queries where CBO brought substantial changes to the query plan and improved performance by more than 30% (shown below). Overall, we observed that the geometric mean speedup for these 16 queries was around 2.2X with Q72 achieving a maximal speedup of 8X.

Conclusion

To recap, this blog post highlighted various aspects of the new cost-based optimizer in Apache Spark 2.2. We discussed details of the statistics collection framework, cardinality propagation for filters and joins, the cost-based optimizations they enable (build side selection and multi-way join reordering), and its performance impact on TPC-DS benchmark queries.

In the past year, we’ve resolved a total of 32 sub-tasks spanning over 50+ patches and 7000+ line of code under the CBO umbrella JIRA SPARK-16026. That said, leveraging cost-based optimizations in distributed databases is an extremely complex problem and this is just a first step in this direction. In future releases, we plan on continuing work in this direction by adding more complex statistics (histograms, count-min sketches, partition-level statistics etc.) and refining our cost formulas.

We are excited about the progress made and hope you will enjoy the improvements. We encourage you to try out the new Cost-Based Optimizer in Apache Spark 2.2!

Read More

You can view our Spark Summit 2017 talk: Cost Based Optimizer in Spark 2.2

Try Databricks for free. Get started today

Originally published at databricks.com on August 31, 2017.

Cost Based Optimizer in Apache Spark 2.2 was originally published in Databricks Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Original authors: Ajay Saini and Joseph Bradley

Developing custom Machine Learning (ML) algorithms in PySpark — the Python API for Apache Spark — can be challenging and laborious. In this blog post, we describe our work to improve PySpark APIs to simplify the development of custom algorithms. Our key improvement reduces hundreds of lines of boilerplate code for persistence (saving and loading models) to a single line of code. These changes are expected to be available in the next Apache Spark release.

Background: PySpark developer APIs

In recent years, Python has become the most popular language for data scientists worldwide, with over a million developers contributing to thousands of open source ML projects. Despite Python’s immense popularity, the developer APIs of Apache Spark MLlib remain Scala-dominated, with all algorithms implemented first in Scala and then made available in Python via wrappers. As a result, it has been very difficult for data scientists to develop ML algorithms in Python without having to write Scala code as well.

This blog post introduces several improvements to PySpark that facilitate the development of custom ML algorithms and 3rd-party ML packages using Python. After introducing the main algorithm APIs in MLlib, we discuss current challenges in building custom ML algorithms on top of PySpark. We then describe our key improvements to PySpark for simplifying such customization.

MLlib algorithm APIs

Before discussing the specific changes to PySpark, it helps to understand the main APIs for ML algorithms in Spark. There are two major types of algorithms: Transformers and Estimators.

Transformers are algorithms that take an input dataset and modify it via a transform() function to produce an output dataset. For example, Binarizer reads an input column of feature values from a dataset, and it outputs a dataset with a new column of 0/1 features based on thresholding the original features.

Estimators are ML algorithms that take a training dataset, use a fit() function to train an ML model, and output that model. That model is itself a Transformer; for models, calling transform() will “transform” the dataset by adding a new column of predictions. Popular examples of Estimators are Logistic Regression and Random Forests.

Users often combine multiple Transformers and Estimators into a data analytics workflow. ML Pipelines provide an API for chaining algorithms, feeding the output of each algorithm into following algorithms. For more details on these types of algorithms, check out the Databricks docs.

Below, we show a simple Pipeline with 2 feature Transformers (Tokenizer, HashingTF) and 1 Estimator (LogisticRegression) from the MLlib guide on Pipelines.

The obstacle: ML Persistence

Let’s say a data scientist wants to extend PySpark to include their own custom Transformer or Estimator. First, the data scientist writes a class that extends either Transformer or Estimator and then implements the corresponding transform() or fit() method in Python. In simple cases, this implementation is straightforward. For example, many feature Transformers can be implemented by using a simple User-Defined Function to add a new column to the input DataFrame.

One critical functionality in MLlib, however, is ML Persistence. Persistence allows users to save models and Pipelines to stable storage, for loading and reusing later or for passing to another team. The API is simple; the following code snippet fits a model using CrossValidator for parameter tuning, saves the fitted model, and loads it back:

val cvModel = cv.fit(training) cvModel.save("myCVModelPath") val sameCVModel = CrossValidatorModel.load("myCVModelPath")

ML Persistence saves models and Pipelines as JSON metadata + Parquet model data, and it can be used to transfer models and Pipelines across Spark clusters, deployments, and teams. For info on persistence, see our blog post and webinar.

Adding support for ML Persistence has traditionally required a Scala implementation. Up until now, the simplest way to implement persistence required the data scientist to implement the algorithm in Scala and write a Python wrapper. Implementing the algorithm in Scala would require knowing both languages, understanding the Java — Python communication interface, and writing duplicate APIs in the two languages.

The solution: Python Persistence mixins

To support Python-only implementations of ML algorithms, we implemented a persistence framework in the PySpark API analogous to the one in the Scala API. With this framework, when implementing a custom Transformer or Estimator in Python, it is no longer necessary to implement the underlying algorithm in Scala. Instead, one can use mixin classes with a custom Transformer or Estimator to enable persistence.

For simple algorithms for which all of the parameters are JSON-serializable (simple types like string, float), the algorithm class can extend the classes DefaultParamsReadable and DefaultParamsWritable (SPARK-21542; code on Github) to enable automatic persistence. (If you are unfamiliar with Params in ML Pipelines, they are standardized ways to specify algorithm options or properties. Refer to the Param section of the MLlib guide for more info.) The default implementation of persistence will allow the custom algorithm to be saved and loaded back within PySpark.

These mixins dramatically reduce the development effort required to create custom ML algorithms on top of PySpark. Persistence functionality that used to take many lines of extra code can now be done in a single line in many cases.

The code snippets below demonstrate the code length of persisting an algorithm with a Scala implementation and a Python wrapper:

And this code snippet demonstrates using these mixins for a Python-only implementation of persistence:

class MyShiftTransformer(UnaryTransformer, DefaultParamsReadable, DefaultParamsWritable):

Adding the mixins DefaultParamsReadable and DefaultParamsWritable to the MyShiftTransformer class allows us to eliminate a lot of code.

For complex algorithms with parameters or data which are not JSON-serializable (complex types like DataFrame), the developer can write custom save() and load() methods in Python. Previously, even with save() and load() implemented, custom Python implementations could not be saved within ML Pipelines. Our fixes (SPARK-17025) correct this issue, allowing smooth integration of custom Python algorithms with the rest of MLlib.

Looking forward

With these improvements, developers will soon be able to write custom machine learning algorithms in Python, use them in Pipelines, and save and load them without touching Scala. We believe this will unblock many developers and encourage further efforts to develop Python-centric Spark Packages for machine learning.

Try Databricks for free. Get started today

Originally published at databricks.com on August 30, 2017.

Developing Custom Machine Learning Algorithms in PySpark was originally published in Databricks Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Older anthologies collated a collection of contributions from various authors around a theme — bounded then as a journal or periodical. Newer anthologies, however, include multiple modals of expressions — digitized now as an ebook or a blog. Both offer an exposition of the subject matter. No matter their form, they provide a single source of focused content.

In this anthology, we have compiled a collection of videos, technical blogs, podcasts, and articles that focus on Apache Spark’s Structured Streaming.

Spark Summit 2017 Keynote: Apache Spark 2.2 and Structured Streaming Demo

Databricks’ Chief Technologist Matei Zaharia thanks the community’s contributions and announces Structured Streaming as ready for production.

Easy, Scalable, Fault-tolerant Stream Processing with Structured Streaming in Apache Spark 2.2

In less than 10 lines of code, you can read streams from Apache Kafka, parse JSON payload data into separate columns, transform it, enrich it by joining with static data and write it out as a table ready for batch or ad-hoc queries. Apache Spark committers and Databricks’ engineers Michael Armbrust and Tathagata Das discuss and demonstrate that with concrete examples.

Also, they explain features that allow event-time based aggregations, arbitrary stateful operations, and automatic state management using event-time watermarks.

Continuous Applications: Evolving Streaming in Apache Spark 2.x

Last year, Databricks’ Chief Technologist Matei Zaharia shared his vision of where Apache Spark streaming is heading: Continuous Applications with Structured Streaming is the next step, he wrote.

Structured Streaming In Apache Spark: A new high-level API for streaming

Databricks’ engineers and Apache Spark committers Matei Zaharia, Tathagata Das, Michael Armbrust and Reynold Xin expound on why streaming applications are difficult to write, and how Structured Streaming addresses all the underlying complexities.

Real-time Streaming ETL with Structured Streaming in Apache Spark 2.1: Part 1 of Scalable Data @ Databricks

Databricks’ engineers Tathagata Das, Michael Armbrust and Tyson Condie show how to do streaming ETL with real-time data at scale.

Working with Complex Data Formats with Structured Streaming in Apache Spark 2.1: Part 2 of Scalable Data @ Databricks

Learn from Databricks engineers and Apache Spark contributors Burak Yavuz, Michael Armbrust, Tathagata Das, and Tyson Condie how to handle complex and nested data formats with Structured Streaming.

Processing Data in Apache Kafka with Structured Streaming in Apache Spark 2.2: Part 3 of Scalable Data @ Databricks

Databricks engineers and Spark contributors Kunal Khamar, Tyson Condie and Michael Armbrust show how easily you can read streams from Apache Kafka using Structured Streaming APIs in Apache Spark 2.2.

Event-time Aggregation and Watermarking in Apache Spark’s Structured Streaming: Part 4 of Scalable Data @ Databricks

How to do event-time aggregations and watermarking using simple Structured Streaming APIs? Databricks engineer and Spark committer Tathagata Das explains how.

Taking Apache Spark’s Structured Streaming to Production: Part 5 of Scalable Data @ Databricks

How do you ensure your Structured Streaming Application is ready for production. Product Manager Bill Chambers and Apache Spark committer Michael Armbrust lay out the vital steps, using simple APIs for alerts and monitoring streaming query states.

Running Streaming Jobs Once a Day For 10x Cost Savings: Part 6 of Scalable Data @ Databricks

Apache Spark contributors Burak Yavuz and Tyson Condie demonstrate how to control and curb costs by using simple APIs such as Run Once trigger feature added to Structured Streaming in Spark 2.2. You get all the benefits of the Catalyst Optimizer incrementalizing your workload and the cost savings of not having an idle cluster lying around.

Real-Time End-to-End Integration with Apache Kafka in Apache Spark’s Structured Streaming

Databricks Senior Solution Architect Sunil Sitaula guides you through an end-to-end integration with Apache Kafka, consuming messages from it, doing simple to complex windowing ETL, and pushing the desired output to various sinks such as memory, console, file, databases, and back to Kafka itself.

Making Apache Spark the Fastest Open Source Streaming Engine

Databricks’ lead on Structured Streaming and Spark committer Michael Armbrust avers why Structured Streaming is the fastest open source engine compared to other streaming engines.

Apache Spark’s Structured Streaming with Amazon Kinesis on Databricks: A quick guide on how to get started with Kinesis Connector

Databricks’ Spark Community Evangelist outlines steps to use AWS Kinesis with Structured Streaming in Apache Spark 2.2 on Databricks Runtime 3.0.

Arbitrary Stateful Aggregations in Structured Streaming in Apache Spark

In this Bay Area Apache Spark Meetup talk, Burak Yavuz, Spark committer and Databricks software engineer, expands on how to use Structured Streaming APIs to maintain stateful aggregations.

Structured Streaming Comes to Apache Spark 2.0

O’Reilly’s Chief Data Scientist Ben Lorica sits down with Michael Armbrust and talks about life and structured streaming.

What Spark’s Structured Streaming Really Means

Ion Pointer (contributor for InfoWorld) advocates why DataFrames are the best choice for Apache Spark Streaming in Spark 2.0, and why structured streaming makes sense.

Apache Spark 2.0 to Introduce New ‘Structured Streaming’ Engine

Datanami sits down with Chief Technologist and Co-founder of Databricks Matei Zaharia to discuss all aspects of Structured Streaming in Apache Spark

What’s Next?

You might want to bookmark this page, as we will update it with part 7 of our series on Structured Streaming. If you want to try some of the notebooks in these assets to explore Spark 2.2’s Structured Streaming features on Databricks Runtime 3.0, you can sign up for a free trial.

Try Databricks for free. Get started today

Originally published at databricks.com on August 24, 2017.

Anthology of Technical Assets on Apache Spark’s Structured Streaming was originally published in Databricks Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Original author: Andrew Chen

This blog post is part of our series of internal engineering blogs on Databricks platform, infrastructure management, integration, tooling, monitoring, and provisioning.

Today, we are excited to announce native Databricks integration in Apache Airflow, a popular open source workflow scheduler. This blog post illustrates how you can set up Airflow and use it to trigger Databricks jobs.

One very popular feature of Databricks’ Unified Analytics Platform (UAP) is the ability to convert a data science notebook directly into production jobs that can be run regularly. While this feature unifies the workflow from exploratory data science to production data engineering, some data engineering jobs can contain complex dependencies that are difficult to capture in notebooks. To support these complex use cases, we provide REST APIs so jobs based on notebooks and libraries can be triggered by external systems. Of these, one of the most common schedulers used by our customers is Airflow. We are happy to share that we have also extended Airflow to support Databricks out of the box.

Airflow Basics

Airflow is a generic workflow scheduler with dependency management. Besides its ability to schedule periodic jobs, Airflow lets you express explicit dependencies between different stages in your data pipeline.

Each ETL pipeline is represented as a directed acyclic graph (DAG) of tasks (not to be mistaken with Spark’s own DAG scheduler and tasks). Dependencies are encoded into the DAG by its edges — for any given edge, the downstream task is only scheduled if the upstream task completed successfully. For example, in the example, DAG below, task B and C will only be triggered after task A completes successfully. Task D will then be triggered when task B and C both complete successfully.

The tasks in Airflow are instances of “operator” class and are implemented as small Python scripts. Since they are simply Python scripts, operators in Airflow can perform many tasks: they can poll for some precondition to be true (also called a sensor) before succeeding, perform ETL directly, or trigger external systems like Databricks.

For more information on Airflow, please take a look at their documentation.

Native Databricks Integration in Airflow

We implemented an Airflow operator called DatabricksSubmitRunOperator, enabling a smoother integration between Airflow and Databricks. Through this operator, we can hit the Databricks Runs Submit API endpoint, which can externally trigger a single run of a jar, python script, or notebook. After making the initial request to submit the run, the operator will continue to poll for the result of the run. When it completes successfully, the operator will return allowing for downstream tasks to run.

We’ve contributed the DatabricksSubmitRunOperator upstream to the open-source Airflow project. However, the integrations will not be cut into a release branch until Airflow 1.9.0 is released. Until then, to use this operator you can install Databricks’ fork of Airflow, which is essentially Airflow version 1.8.1 with our DatabricksSubmitRunOperator patch applied.

pip install --upgrade "git+git://github.com/databricks/incubator-airflow.git@1.8.1-db1#egg=apache-airflow[databricks]"

Airflow with Databricks Tutorial

In this tutorial, we’ll set up a toy Airflow 1.8.1 deployment which runs on your local machine and also deploy an example DAG which triggers runs in Databricks.

The first thing we will do is initialize the sqlite database. Airflow will use it to track miscellaneous metadata. In a production Airflow deployment, you’ll want to edit the configuration to point Airflow to a MySQL or Postgres database but for our toy example, we’ll simply use the default sqlite database. To perform the initialization run:

airflow initdb

The SQLite database and default configuration for your Airflow deployment will be initialized in ~/airflow.

In the next step, we’ll write a DAG that runs two Databricks jobs with one linear dependency. The first Databricks job will trigger a notebook located at /Users/airflow@example.com/PrepareData, and the second will run a jar located at dbfs:/lib/etl-0.1.jar. To save time, we’ve already gone ahead and written the DAG for you here.

From a mile high view, the script DAG essentially constructs two DatabricksSubmitRunOperator tasks and then sets the dependency at the end with the set_dowstream method. A skeleton version of the code looks something like this:

notebook_task = DatabricksSubmitRunOperator(
    task_id='notebook_task',
    …)

spark_jar_task = DatabricksSubmitRunOperator(
    task_id='spark_jar_task',
    …)
notebook_task.set_downstream(spark_jar_task)

In reality, there are some other details we need to fill in to get a working DAG file. The first step is to set some default arguments which will be applied to each task in our DAG.

args = {
    'owner': 'airflow',
    'email': ['airflow@example.com'],
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2)
}

The two interesting arguments here are depends_on_past and start_date. If depends_on_past is true, it signals Airflow that a task should not be triggered unless the previous instance of a task completed successfully. The start_date argument determines when the first task instance will be scheduled.

The next section of our DAG script actually instantiates the DAG.

dag = DAG(
    dag_id='example_databricks_operator', default_args=args,
    schedule_interval='@daily')

In this DAG, we give it a unique ID, attach the default arguments we declared earlier, and give it a daily schedule. Next, we’ll specify the specifications of the cluster that will run our tasks.

new_cluster = {
    'spark_version': '2.1.0-db3-scala2.11',
    'node_type_id': 'r3.xlarge',
    'aws_attributes': {
        'availability': 'ON_DEMAND'
    },
    'num_workers': 8
}

The schema of this specification matches the new cluster field of the Runs Submit endpoint. For your example DAG, you may want to decrease the number of workers or change the instance size to something smaller.

Finally, we’ll instantiate the DatabricksSubmitRunOperator and register it with our DAG.

notebook_task_params = {
    'new_cluster': new_cluster,
    'notebook_task': {
        'notebook_path': '/Users/airflow@example.com/PrepareData',
    },
}

# Example of using the JSON parameter to initialize the operator.
notebook_task = DatabricksSubmitRunOperator(
    task_id='notebook_task',
    dag=dag,
    json=notebook_task_params)

In this piece of code, the JSON parameter takes a python dictionary that matches the Runs Submit endpoint.

To add another task downstream of this one, we do instantiate the DatabricksSubmitRunOperator again and use the special set_downstream method on the notebook_task operator instance to register the dependency.

# Example of using the named parameters of DatabricksSubmitRunOperator
# to initialize the operator.
spark_jar_task = DatabricksSubmitRunOperator(
    task_id='spark_jar_task',
    dag=dag,
    new_cluster=new_cluster,
    spark_jar_task={
        'main_class_name': 'com.example.ProcessData'
    },
    libraries=[
        {
            'jar': 'dbfs:/lib/etl-0.1.jar'
        }
    ]
)

notebook_task.set_downstream(spark_jar_task)

This task runs a jar located at dbfs:/lib/etl-0.1.jar.

Notice that in the notebook_task, we used the JSON parameter to specify the full specification for the submit run endpoint and that in the spark_jar_task, we flattened the top level keys of the submit run endpoint into parameters for the DatabricksSubmitRunOperator. Although both ways of instantiating the operator are equivalent, the latter method does not allow you to use any new top level fields like spark_python_task or spark_submit_task. For more detailed information about the full API of DatabricksSubmitRunOperator, please look at the documentation here.

Now that we have our DAG, to install it in Airflow create a directory in ~/airflow called ~/airflow/dags and copy the DAG into that directory.

At this point, Airflow should be able to pick up the DAG.

$ airflow list_dags                                                           [10:27:13]
[2017-07-06 10:27:23,868] {__init__.py:57} INFO - Using executor SequentialExecutor
[2017-07-06 10:27:24,238] {models.py:168} INFO - Filling up the DagBag from /Users/andrew/airflow/dags


-------------------------------------------------------------------
DAGS
-------------------------------------------------------------------
example_bash_operator
example_branch_dop_operator_v3
example_branch_operator
example_databricks_operator

We can also visualize the DAG in the web UI. To start it up, run airflow webserver and connect to localhost:8080. Clicking into the “example_databricks_operator,” you’ll see many visualizations of your DAG. Here is the example:

At this point, a careful observer might also notice that we don’t specify information such as the hostname, username, and password to a Databricks shard anywhere in our DAG. To configure this we use the connection primitive of Airflow that allows us to reference credentials stored in a database from our DAG. By default, all DatabricksSubmitRunOperator set the databricks_conn_id parameter to “databricks_default,” so for our DAG, we’ll have to add a connection with the ID “databricks_default.”

The easiest way to do this is through the web UI. Clicking into the “Admin” on the top and then “Connections” in the dropdown will show you all your current connections. For our use case, we’ll add a connection for “databricks_default.” The final connection should look something like this:

Now that we have everything set up for our DAG, it’s time to test each task. To do this for the notebook_task we would run, airflow test example_databricks_operator notebook_task 2017-07-01 and for the spark_jar_task we would run airflow test example_databricks_operator spark_jar_task 2017-07-01. To run the DAG on a schedule, you would invoke the scheduler daemon process with the command airflow scheduler.

If everything goes well, after starting the scheduler, you should be able to see backfilled runs of your DAG start to run in the web UI.

Next Steps

In conclusion, this blog post provides an easy example of setting up Airflow integration with Databricks. It demonstrates how Databricks extension to and integration with Airflow allows access via Databricks Runs Submit API to invoke computation on the Databricks platform. For more detailed instructions on how to set up a production Airflow deployment, please look at the official Airflow documentation.

Also, if you want to try this tutorial on Databricks, sign up for a free trial today.

This article was originally published on the Databricks Blog.

Integrating Apache Airflow with Databricks was originally published in Databricks Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Original authors: Weiluo Ren, Yu Peng and Xiangrui Meng

This blog post is part of our series of internal engineering blogs on Databricks platform, infrastructure management, tooling, monitoring, and provisioning.

We love logs at Databricks. And we want to provide users the best experience to access Apache Spark logs. On Databricks, Spark logs are available via built-in Spark UI for clusters. Databricks also supports delivering cluster logs to customers’ chosen destination.

While we care about the ease of accessing Spark logs, we pay more attention to data security issues that appear in logging. In this blog post, we discuss a specific data security topic on Spark logs: redacting credentials in logs.

Credentials in Apache Spark Logs

Spark can load data from various data sources, where users may need to provide credentials for accessing data. For instance, a Spark job may read a dataset from S3 using AWS access and secret access keys:

spark.read.parquet("s3a://AKIAABCDEFGHIJKLMNOP:ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmn@test-bucket/test/path")

Anyone who obtained a copy of the keys would have the same access to the corresponding AWS resources. Unfortunately, the S3 URI that contains the credentials may appear in Spark logs in plain text form:

17/01/01 00:00:00 INFO FileScanRDD: Reading File path: s3a://AKIAABCDEFGHIJKLMNOP:ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmn@test-bucket/test/path, range: ...

This could lead to serious data security issues if logs do not receive the same level of protection as credentials. And it is hard for users to control or watch what content are logged by Spark and third-party packages.

To enhance data security, Databricks redacts credentials from Spark logs at logging time, before log messages are written to the disk. For example, the above log message would appear in redacted form on Databricks:

17/01/01 00:00:00 INFO FileScanRDD: Reading File path: s3a://REDACTED_AWS_ACCESS_KEY:REDACTED_AWS_SECRET_ACCESS_KEY@test-bucket/test/path, range: ...

In the following sections, we’ll discuss how we attempt to properly identify credentials and redact them at logging time. Although we’ll use AWS keys as examples for this blog post, Databricks also redacts other types of credentials.

Identifying AWS Credentials

Before each message gets logged, Databricks scans the whole string to identify any possible credential and redacts them. Because redaction is done at logging time, we cannot afford an overly sophisticated redaction procedure that may hurt the performance. With complexity constraints in mind, log redaction at Databricks is developed based on regular expressions (regexes).

AWS access keys consist of 20-character upper case alphanumeric strings starting with “AKIA,” which can be captured by the following regex:

(?<![A-Z0-9])AKIA([A-Z0-9]){16}(?![A-Z0-9])

For secret keys, there is no explicit pattern except that they are 40-character base64 strings. AWS provided a regex for access keys in the blog, A Safer Way to Distribute AWS Credentials to EC2, which is a good start:

(?<![A-Za-z0-9/+=])[A-Za-z0-9/+=]{40}(?![A-Za-z0-9/+=])

However, we found this regex insufficient to capture all secret keys. Some secret keys contains special characters such as the forward slash “/”, which needs to be URL-encoded (e.g., “/” encoded as “%2F”) to appear in a file path. Therefore, to identify secret keys in Spark logs, we also need to take into account the URL encoding on special characters. Moreover, both users and Spark may escape special characters in the URI, so in some cases special characters may be encoded twice. For instance, instead of “%2F”, “/” may be encoded and logged as “%252F”. We improved the regular expression accordingly as follows, which matches the secret keys not only in code but also in logs:

(?<![A-Za-z0-9/+])([A-Za-z0-9/+=]|%2F|%2B|%3D|%252F|%252B|%253D){40}(?![A-Za-z0-9/+=])

False Positives and Entropy Thresholding

One issue of using the regular expression above is that it is not specific enough for secret keys, which could cause a number of false positives when applying to all Spark logs. For instance, it could also match hashcodes, paths or even class names that happen to be 40 characters long. To reduce the false positive rate without missing real secret keys, more checks are needed after regular expression matching.

One difference between secret keys and informative false positives is that the secret keys are randomly generated. One way to differentiate secret keys from those false positives is to measure how random they are. Following this idea, we use entropy to determine how likely a matched string is indeed a secret key that was randomly generated.

In information theory, entropy, or Shannon entropy, is the expected value of the information contained in each message (Wiki page). This concept can be applied to strings to measure how random they are. We expect randomly generated strings to have higher entropy than meaningful words with the same length.

After calculating entropy, we set a threshold to decide whether to redact a given string. The threshold is based on the empirical distribution of AWS secret keys and those of known false positive types. One type of false positives is the SHA-1 hashcode, which is 40 characters long and randomly generated. However, the SHA-1 hashcode only consists of hexadecimal digits. Roughly speaking, with the same length, the more distinct characters a string have, the higher its entropy would be. Therefore, the entropy of SHA-1 hashcodes would usually be smaller than that of AWS secret keys, which are generated from base64 character set. By comparing the empirical distribution of entropy of two datasets consisting of AWS secret keys and SHA-1 hashcodes respectively (see figure below), it turns out that there’s a clear gap between those two distributions and we can choose a threshold accordingly.

Logging-time Redaction with log4j

Logs generated by Apache Spark jobs at Databricks are mostly handled by the logging service log4j. Databricks customizes log4j appenders, for example, console appender and rolling file appender to redact log strings before they arrive at the disk. Note that the redaction is applied to not only normal messages, but also stack traces when exceptions are thrown.

One interesting lesson learned is that since logging is so pervasive at Databricks, we need to minimize the dependency of customized appenders. Otherwise there could be a circle dependency issue.

Using IAM Roles Instead of Keys

In this blog post, we shared our experience on redacting credentials at logging time. On Databricks this security feature is turned on automatically[1]. Outside of Databricks, users can implement the methods mentioned to improve data security in Spark logs.

However, rather than doing redaction at logging time, the best security practice is to simply not use credentials in the first place. Databricks encourages users to use IAM roles to access AWS resources. Users can launch Databricks clusters with IAM roles that grant the workers the corresponding permissions. With IAM-enabled clusters, users no longer need to embed their AWS keys in notebooks to access data, and hence keys won’t appear in logs.

Similarly, in Databricks File System (DBFS), users can mount their S3 bucket with IAM roles. Users can create clusters with corresponding IAM roles to access data directly without keys.

This article was originally published on the Databricks Blog.

1. see https://docs.databricks.com/user-guide/security.html for details

Entropy-based Log Redaction for Apache Spark on Databricks was originally published in Databricks Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Original authors: Eric Liang and Aaron Davidson.

This blog post is part of our series of internal engineering blogs on Databricks platform, infrastructure management, integration, tooling, monitoring, and provisioning.

At Databricks engineering, we are avid fans of Kubernetes. Much of our platform infrastructure runs within Kubernetes, whether in AWS cloud or more regulated environments.

However, we have found that Kubernetes alone is not enough for managing a complex service infrastructure, which may encompass both resources created in Kubernetes (e.g., pods, services) and external resources such as IAM roles. Management complexity comes from (1) need for visibility into the current state of the infrastructure (2) reasoning about how to make changes to the infrastructure.

To help reduce management complexity, it is desirable for infrastructure to be declarative (i.e., described by a set of configuration files or templates). The first advantage is that one can easily inspect the target state of infrastructure by reading the configuration files. Second, since the infrastructure is entirely described by the files, changes can be proposed, reviewed, and applied as part of a standard software development workflow.

What’s Missing?

Kubernetes YAML configuration files already implement declarative updates to objects in Kubernetes. The user only need edit an object’s YAML file and then run $ kubectl apply -f theobject.yaml to sync the changes to Kubernetes. These YAML files may be checked into source control, and if needed the user can query Kubernetes to inspect the difference between the live version and the local file.

However, one runs into pain points when attempting to apply this methodology to a larger production environment:

1. Common structures cannot be shared across YAML files. You typically set up a service not once but several times, perhaps in {dev, staging, prod} environments, across different geographic regions such as {us-west, us-east, asia-pacific}, and Cloud providers {AWS, Azure, GCE}.
2. YAML files must often reference metadata about externally defined entities such as relational databases, networking configurations, or SSL certificates.
3. Complex multi-tier service deployments may require the composition of many distinct resources or YAML files. Updating a deployment composed of many such files becomes burdensome.

In this blog post, we describe how we use Google’s Jsonnet configuration language to solve these problems, walk through an example of templatizing a service deployment using Jsonnet, and propose a Jsonnet style guide for the infrastructure templating use case. We’ve found that Jsonnet is easy to get started with and scales well to complex use cases.

Jsonnet Usage at Databricks

Late 2015 we started experimenting with Jsonnet as part of the effort for Databricks Community Edition, our free tier Apache Spark service for education. Since then, Jsonnet has exploded in popularity within Databricks engineering, with over 40,000 lines of Jsonnet in 1000+ distinct files checked into our main development repository. These templates expand to hundreds of thousands of lines of raw YAML once materialized.

Teams at Databricks currently use Jsonnet to manage configurations for Kubernetes resources (including internal configurations of services running on Kubernetes), AWS CloudFormation, Terraform, Databricks Jobs, and also for tasks such as TLS cert management and defining Prometheus alerts. Over the past two years, Jsonnet has grown to become the de-facto standard configuration language within engineering.

Jsonnet Basics

Jsonnet is a configuration language that helps you define JSON data. The basic idea is that some JSON fields may be left as variables or expressions that are evaluated at compilation time. For example, the JSON object {"count": 4} may be expressed as {count: 2 + 2} in Jsonnet. You can also declare hidden fields with “::” that can be referenced during compilation, e.g. {x:: 2, y:: 2, count: $.x + $.y} also evaluates to {"count": 4}.

Jsonnet objects may be subclassed by concatenating (“+”) objects together to override field values, e.g. suppose we define the following.

local base = {
   x:: error "you must define the value of 'x'",
   y:: 2,  // this field has a default value
   count: $.x + $.y
};

Then, the Jsonnet expression (base + {x:: 10}) will compile to {"count": 12}. In fact, the Jsonnet compiler requires you to override x since the default value raises an error. You can, therefore, think of base as defining an abstract base class in Jsonnet.

Jsonnet compilation is completely deterministic and cannot perform external I/O, making it ideal for defining configuration. We’ve found that Jsonnet strikes the right balance between restrictiveness and flexibility — previously we generated configurations using Scala code, which erred too much on the side of flexibility, leading to many headaches.

At Databricks, we extended our kubectl tool (dubbed kubecfg) so that it can take Jsonnet files directly as arguments. Internally it compiles the Jsonnet to plain JSON / YAML before sending the data to Kubernetes. Kubernetes then creates or updates objects as needed based on the uploaded configuration.

Composing Kubernetes Objects with Jsonnet

To better understand how Jsonnet can be used with Kubernetes, let’s consider the task of deploying an idealized [1] single-tenant “Databricks platform” for an enterprise customer. Here we want to create two distinct but related service deployments: a Webapp (for the interactive workspace) and Cluster manager (for managing Spark clusters). In addition, the services need access to an AWS RDS database to store persistent data.

[1]: Note that we don’t actually create individual Jsonnet files for each customer in reality — that would create a frightening number of Jsonnet files and would in itself be a problem.

Template for Defining Kubernetes Deployments

For these examples, we’ll be using service-deployment.jsonnet.TEMPLATE, a simplified version of one of our internal base templates that define a Kubernetes service and deployment together as a pair. The instantiated service and deployment together constitute a standalone “production service” in Kubernetes that can receive network traffic. Note that the template has two required arguments in addition to several optional arguments, including optional config passed to the service binary itself:

service-deployment.jsonnet.TEMPLATE:

// Template for a Kubernetes (service, deployment) pair.
{
  // Required arguments for this template
  serviceName:: error "serviceName must be specified",
  dockerImage:: error "dockerImage must be specified",

  // Optional arguments for this template.
  serviceConf:: {}
  ...
}

Example 1: One File for Each Service Deployment

Given service-deployment.jsonnet.TEMPLATE, the simplest option we have is to create separate Jsonnet files for the Webapp and Cluster manager. In each file we must import the base template, subclass it, and fill in the required parameters.

We specify the service name, the Docker image containing the service binary, and some service specific configurations including the RDS address. In this example, the RDS address is hard-coded, but later on we’ll show how to import that metadata from e.g. the output file of running a CloudFormation script.

The following file describes the manager service (here service takes on its standard meaning, we’ll refer to Kubernetes services explicitly as such going on). We also construct service-specific configurations within the Jsonnet template via the serviceConf:: field, which the template passes to the pod as an environment variable. We’ve found it useful to unify service and Kubernetes configs in this way:

simple/foocorp-manager.jsonnet:

local serviceDeployment = import "../service-deployment.jsonnet.TEMPLATE";

// The cluster manager deployment for foocorp.
serviceDeployment + {
  serviceName:: "foocorp-manager",
  dockerImage:: "manager:2.42-rc1",
  serviceConf:: {
    customerName: "foocorp",
    database: "user-db.databricks.us-west-2.rds.amazonaws.com",
  },
}

For the webapp service to create clusters, it must specify the cluster manager’s Kubernetes DNS address, which can be determined ahead of time from the Kubernetes service name:

simple/foocorp-webapp.jsonnet:

local serviceDeployment = import "../service-deployment.jsonnet.TEMPLATE";

// The webapp deployment for foocorp.
serviceDeployment + {
  serviceName:: "foocorp-webapp",
  dockerImage:: "webapp:2.42-rc1",
  serviceConf:: {
    customerName: "foocorp",
    database: "user-db.databricks.us-west-2.rds.amazonaws.com",
    managerAddress: "foocorp-manager.prod.svc.cluster.local",
  },
}

If you have the example code repo cloned, you can view the materialized outputs of these templates by running the jsonnet compiler on the files, e.g.:

$ jsonnet examples/databricks/simple/foocorp-webapp.jsonnet

So what has this gotten us? We’ve managed to at least remove some of the standard Kubernetes boilerplate around defining services, reducing each deployment definition from over 100 lines to about 10. However, we can do better — there are still duplicated parameters which would make this pattern hard to maintain if there were many distinct customers or more services required per customer.

Example 2: Composing Together Deployments in a Single File

Since both the Webapp and Cluster manager services are deployed together as a unit or shard per customer, it makes sense that there should just be one file per customer describing their unique requirements. It is indeed possible to define a single template, shard.jsonnet.TEMPLATE, that composes together both the Webapp and Cluster manager deployments without duplication of params.

The trick here is that the template defines a Kubernetes “List” object, which can include multiple Kubernetes resources within one JSON object. We merge the sub-lists produced by the service-deployment templates using the Jsonnet standard library function std.flattenArrays:

shard-v1/shard.jsonnet.TEMPLATE:

local serviceDeployment = import "../service-deployment.jsonnet.TEMPLATE";

{
  // Required arguments
  customerName:: error "customerName must be defined",
  release:: error "release must be defined",

  // Optional arguments
  commonConf:: {
    customerName: $.customerName,
    database: "user-db.databricks.us-west-2.rds.amazonaws.com",
  },

  local webapp = serviceDeployment + ...
  local manager = serviceDeployment + ...

  kind: "List",
  items: std.flattenArrays([webapp, manager]),
}

Now we can conveniently define FooCorp’s entire deployment without any duplicated data, and using only one file:

shard-v1/foocorp-shard.jsonnet:

local shardTemplate = import "shard.jsonnet.TEMPLATE";

shardTemplate + {
  customerName:: "foocorp",
  release:: "2.42-rc1",
}

Example 3: Subclassing Templates for Multiple Environments

To see the flexibility of Jsonnet, consider how you might subclass the templates defined in the above examples to create shards that run in the development environment. In dev, deployments should use the dev database instead of the production one, and we also want to run with the latest bleeding-edge Docker images. To do this, we can further subclass the shard.jsonnet.TEMPLATE to different environments. This makes it possible to create specific production or development shards from these specialized templates:

To make the interface exposed by shard.jsonnet.TEMPLATE more explicit, instead of exporting an object we will export a function newShard(name, release, env), which can be called to construct the shard object. The env object encapsulates the differences between environments (e.g. database URL).

shard-v2/shard.jsonnet.TEMPLATE:

local serviceDeployment = import "../service-deployment.jsonnet.TEMPLATE";

local newShard(customerName, release, env) = {
  local commonConf = {
    customerName: customerName,
    database: env.database,
  },

  local webapp = serviceDeployment + {
    serviceName:: customerName + "-webapp",
    dockerImage:: "webapp:" + release,
    serviceConf:: commonConf + {
      managerAddress: customerName + "-manager.prod.svc.cluster.local",
    },
  },

  local manager = serviceDeployment + {
    serviceName:: customerName + "-manager",
    dockerImage:: "manager:" + release,
    serviceConf:: commonConf,
  },

  apiVersion: "v1",
  kind: "List",
  items: std.flattenArrays([webapp.items, manager.items]),
};

// Export the function as a constructor for shards
{
  newShard:: newShard,
}

The dev shard template now just needs to fill in the required env for dev and specify the bleeding-edge release by default. It obtains the env by importing a plain JSON file containing the needed database metadata:

shard-v2/dev-shard.jsonnet.TEMPLATE:

local shardTemplate = import "shard.jsonnet.TEMPLATE";
local devEnv = import "dev-env.json";

local newDevShard(shardName, release="bleeding-edge") = (
  shardTemplate.newShard(shardName, release, env=devEnv)
);

{
  newDevShard:: newDevShard,
}

The full example code for the prod and dev shard templates can be found here: https://github.com/databricks/jsonnet-style-guide/tree/master/examples/databricks/shard-v2

Note that the use of functions for subclassing is optional — we could have instead derived dev-shard.jsonnet.TEMPLATE and prod-shard.jsonnet.TEMPLATE using only field overrides. However, in our experience, the use of ad-hoc field overrides, though powerful, tends to lead to more brittle templates. We’ve found it useful to establish best practices to limit the use of such constructs in larger templates.

Jsonnet Style Guide

In the above examples, we saw how Jsonnet templates can be used to remove duplication of configuration data between Kubernetes objects, compose multiple deployments together, and reference external entities. This greatly simplifies infrastructure management. In large projects, however, templates themselves may become a source of complexity. We’ve found this to be manageable with some best practices and investment in tooling around Jsonnet:

Here are a few Jsonnet best practices we’ve found:

1. For large templates (>10 parameters), avoid directly overriding internal fields when subclassing. Rather, define explicit constructors for templates using Jsonnet functions, which helps with readability and encourages modularity.
2. Check your Jsonnet configurations into source control.
   - Add pre-commit tests to make sure all checked in templates compile. This avoids inadvertent breakages when common templates are updated. You can also create “unit tests” using Jsonnet assert expressions that assert invariants over template variables.
   - Consider also checking in the Jsonnet materialized JSON / YAML to make changes more visible during code review. Because common Jsonnet templates may be imported by many files, changing one file can affect the output of many. Fortunately, unintended changes can be detected at compile time if the materialized YAML is also included with the Jsonnet changes.
   - Refactor often — there’s no risk of breakage. Since Jsonnet templates compile to concrete JSON / YAML files, it’s possible to check the before and after outputs to ensure the refactoring was correct (i.e. zero net materialized diff).
3. Push as much configuration to Jsonnet as possible. It may be tempting to introduce logic into services based on some environment flag (e.g. dev vs prod). We’ve found this to be an anti-pattern — such configuration should be determined at template compile time rather than at runtime. This principle of hermetic configuration helps prevent surprises when services are deployed to new environments.

For more detailed recommendations, check out our newly posted Jsonnet style guide. We also want to welcome any comments or contributions there from the community.

Conclusion

In this blog post, we’ve shared our experience at Databricks using Jsonnet to simplify infrastructure management. As a unifying templating language that enables composition across a broad range of infrastructure and services, Jsonnet brings us one step closer to a totally declarative infrastructure. We’ve found Jsonnet easy to use and flexible enough to serve as our primary configuration language, and suggest you try it too!

If you found this topic interesting, watch out for future blog posts on the tooling we’ve built around Jsonnet and Kubernetes. We’re also hiring for Cloud platform, infrastructure, and Databricks Serverless.

This article was originally published on the Databricks Blog.

Declarative Infrastructure with the Jsonnet Templating Language was originally published in Databricks Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.