What is TensorFrames? TensorFlow + Apache Spark
Tags: opensource, beginners, tensorflow, apachespark
Published: 2019-03-25 14:55:49

---
title: What is TensorFrames?  TensorFlow + Apache Spark
published: true
description: Passionate about AI? wonder what TensorFrames is and how it’s related to TensorFlow? read this!
tags: opensource , beginners , tensorflow , apachespark
cover_image: https://images.unsplash.com/photo-1540740887863-4fa5222f3183?ixlib=rb-1.2.1&ixid=eyJhcHBfaWQiOjEyMDd9&auto=format&fit=crop&w=1950&q=80
---

* * *

**First thing first, what is TensorFrames?**

TensorFrames is an open source created by Apache Spark contributers. It's functions and parameters are neamed the same as in the TensorFlow framework. Under the hood it is an Apache Spark DSL (domain-specific language) wrapper for Apache Spark DataFrames. It allow us to manipulate the DataFrames with TensorFlow functionality. And no, it is **not** [pandas DataFrame](https://medium.com/r/?url=https%3A%2F%2Fwww.tutorialspoint.com%2Fpython_pandas%2Fpython_pandas_dataframe.htm), it is based on Apache Spark DataFrame. 

**..but wait, what is TensorFlow (TF)?**

TensorFlow is an open-source software library for dataflow and differentiable programming across a range of tasks. It is a symbolic math library and is also used for machine learning applications such as neural networks.

![](https://cdn-images-1.medium.com/max/1600/1*f5ySeQqn5E8SpRBZXKcvCQ.jpeg)

**..and Apache Spark?**

Apache Spark is an open-source distributed general-purpose cluster-computing framework.

![](https://cdn-images-1.medium.com/max/1600/0*YgOQ8BsHBSg4BAMJ.png)



#### A word about scale

Today when we mention scale, we usually talk about two options; scale horizontally, and scaling vertically. 

*   **Horizontal scale** – add additional machines with more or less the same computing power 
*   **Vertical scale** - adding more resources to machine/s we are currently working with. It can be a processor upgraded from a CPU to GPU, more memory (RAM), and etc.

With TensorFrames, we can do both, more processor computing power, and more machines. Where with only TensorFlow we would usually focus on adding more power through scaling vertically, now with Apache Spark support, we can scale both vertically and horizontally. But, how do we know how much of each we actually need? To answer this question, we need to understand the full usage of our applications and plan accordingly.

For each change, like adding a machine or upgrading from CPU to GPU, we have _downtime._ In the cloud, resizing a cluster or adding more compute power, is a matter of minutes, versus on-prem where we need to deal with adding new machines and upgrading machines processors, this can take days, and sometimes months. 

So, A more flexible solution is the public cloud. 

In the picture below, scale horizontally is the X-axis where scale vertically is the Y-axis.

![](https://cdn-images-1.medium.com/max/1600/1*ANnCcyS8zH3FBi0n7T7eFA.png)**Slide from Tim Hunter presentation at Apache Spark conf



#### Before jumping to the functions, let’s understand some important TensorFlow vocabulary:

#### Tensor

A statically typed multi-dimensional array whose elements are of a generic type.

#### GraphDef

`Graph` or `Computional Graph` is the core concept of TensorFlow to present computation. When we use TensorFlow, we first create our own `Computation Graph` and pass the `Graph` to TensorFlow. `GraphDf` is the serialized version of `Graph`.

#### Operation

A Graph node that performs computation on Tensors. An Operation is a node in a `Graph` that takes zero or more `Tensors` (produced by other Operations in the Graph) as input and produces zero or more `Tensor`s as output.

#### Identity

`tf.identity` is used when we want to explicitly transport tensor between devices (like, from GPU to a CPU). The operation adds nodes to the graph, which makes a copy when the devices of the input and the output are different.

#### Constant

A constant has the following arguments which can be tweaked as required to get the desired function. It the same as a variable, but its value can’t be changed. Constant can be: 

*   `value`: A constant value (or list) of output type `dtype`.
*   `dtype`: The type of the elements of the resulting tensor.
*   `shape`: Optional dimensions of resulting tensor.
*   `name`: Optional name for the tensor.
*   `verify_shape`: Boolean that enables verification of a shape of values.

#### Placeholders

Allocate storage for data (such as for image pixel data during a feed). Initial values are not required (but can be set, see `tf.placeholder_with_default`). Versus variables, where you need to declare the initial value. \

#### Some Apache Spark Vocabulary 

#### Dataframe 

This is a _distributed_ collection of data organized into named columns that provide operations to filter, group, or compute aggregates. Dataframe data is often distributed across multiple machines. It can be in memory data or on disk.

#### RelationalGroupedDataset

A set of methods for aggregations on a `DataFrame`, created by [groupBy](https://medium.com/r/?url=http%3A%2F%2Fmaxpumperla.com%2Fjava-book%2Fapi%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2FDataset.html%23groupBy%28col1%3AString%2Ccols%3AString*%29%3Aorg.apache.spark.sql.RelationalGroupedDataset), [cube](https://medium.com/r/?url=http%3A%2F%2Fmaxpumperla.com%2Fjava-book%2Fapi%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2FDataset.html%23cube%28col1%3AString%2Ccols%3AString*%29%3Aorg.apache.spark.sql.RelationalGroupedDataset) or [rollup](https://medium.com/r/?url=http%3A%2F%2Fmaxpumperla.com%2Fjava-book%2Fapi%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2FDataset.html%23rollup%28col1%3AString%2Ccols%3AString*%29%3Aorg.apache.spark.sql.RelationalGroupedDataset).

The main method is the `agg` function, which has multiple variants. This class also contains some first-order statistics such as `mean`, `sum` for convenience.

Now that we understand the terminology better, let’s look at the functionality.

### The Functionality — TensorFlow version 0.6.0

Apache Spark is known for being an analytics platform for data at scale, together with TensorFlow, we get TensorFrames which have three categories of data manipulations: 

*   Mapping
*   Reducing
*   Aggregation


Let's understand each functionality.

<pre name="73cd" id="73cd" class="graf graf--pre graf-after--p">  -1- Mapping </pre>

Mapping operations transform and/or adds columns to a given dataframe. 

Each functionality is accessed through two API, one which receives Operation and the other which receives DataFrame, GraphDef, and ShapeDescription.

Exposed API: 

 {% gist https://gist.github.com/adipola/bbd41335b59e907086e72a08e084adae %}
    

#### MapRows

For the user, this is the function that will be more often in use, since there is no direct request to create the GraphDef and ShapeDescription object. This way is more readable for experienced TensorFlow developers: 


`def mapRows(o0: Operation, os: Operation*): DataFrame`

mapRows receives two parameters, operation, and operation* which means the second operation can be a collection of operations. Later it turns them into a sequence and translates it into a graph, it creates the ShapeDiscription out of the graph and sends it with the DataFrame to an internal function. Where it transforms the distributed data row by row according to the transformations given in the graph. All input in the graph should be filled with some data from the given DataFrame or constants. Meaning, we can’t use null. At the end the function returns a new DataFrame with the new schema, the schema will contain the original schema **plus** new columns that correspond to the graph output. ShapeDiscription provides the shape of the output, it is used, behain the sceans, for optimization and going around limitations of the kernel.



#### MapBlock

Performs a similar task as `MapRows`, however, since it is optimized for compact, it applies the graph transformers in blocks of data and not row by row. 

The often more used function is:

`def mapBlocks(o0: Operation, os: Operation*): DataFrame`

Code example: We create _val df_, which is of type DataFrame, with two rows, one contains value 1.0 and the second data row contain value 2.0\. The column name is x.

_val x_ is a declaration of the placeholder for the output, _y_ is the identity for transporting tensors from CPU to GPU or from machine to machine, it received _val x_ as it's value. 

 _z_ is the computation function itself. Here, `df.MapBlock` functions gets two operations, _y_ and _z_, and retunes a new DataFrame named _df2_ with extra column _z_. Column _z_ is the sum of _x+x._ In the output, _column x_ is the original value, _column y_ is the identity value and _column z_ is the output of the graph.


{% gist https://gist.github.com/adipola/d925d7c3ad4d4fdb5bd328daec02b743 %}

#### MapBlocksTrimmed

This is the same as `MapBlock` , BUT, it drops the original DataFrame columns from the result DataFrame. Meaning the output DataFrame will contain only the calculated columns. 

Let’s look at:

`def mapBlocksTrimmed(o0: Operation, os: Operation*): DataFrame`

Code example: we create a DataFrame named _df_ with two rows with values 3.0 and 4.0 . **Notice** that we create a constant named _out_ with value 1.0 and 2.0,this constant is the TensorFrame dsl functionality that mimics the TensorFlow functionality. Then we call `df.MapBlocksTrimmed`. The output schema will only contain the result column, which is named “out” and in our case will only hold the constant values which are 1.0 and 2.0 .

{% gist https://gist.github.com/adipola/daeb530f2654a26d385756e3e528b27d %}



**Important Note**  in the first line of code we import TesnorFrames dsl and we name it to tf, which stands for TensorFlow, we do it since this is how TesnorFlow users used to work with it and we are adhering to the best practices of TensorFlow.

<iframe data-width="435" data-height="429" width="435" height="429" src="/media/a428c9f55b6563fe35727265cfffcb09?postId=c9773a3cd368" data-media-id="a428c9f55b6563fe35727265cfffcb09" data-thumbnail="https://i.embed.ly/1/image?url=https%3A%2F%2Fmedia.giphy.com%2Fmedia%2F13jVk3cL53v4LS%2Fgiphy.gif&amp;key=a19fcc184b9711e1b4764040d3dc5c07" allowfullscreen="" frameborder="0"></iframe>

<pre name="3290" id="3290" class="graf graf--pre graf-after--p"> -2- Reducing </pre>

Reduction operations coalesce a pair or a collection of rows and transform them into a single row, it repeats the same operation until there is one row left. Under the hood, TensorFrames minimizes the data transfer between computers by reducing all the rows on each computer first and then sending the remainder over the network to perform the last reductions.

The transforms function must be classified as [morphism](https://medium.com/r/?url=https%3A%2F%2Fen.wikipedia.org%2Fwiki%2FMorphism): the order in which they are done should not matter. In mathematical terms, given some function `f` and some function inputs `a`, `b`, `c`, the following must hold:

`f(f(a, b), c) == f(a, f(b, c))`

![](https://cdn-images-1.medium.com/max/1600/1*HTqayi_15eTfnSdlQfE5ng.png)Map reduce schema by [Christopher Scherb](https://medium.com/r/?url=https%3A%2F%2Fwww.researchgate.net%2Fprofile%2FChristopher_Scherb)

The reduce functionality API, same as the rest, we have 2 API for each functionality, the one which receives Operation is more intuitive, however, in TensorFlow there is no direct reduce rows operation, instead, there are many reduce operations like `tf.math.reduce_sum` and `tf.reduce_sum` .

{% gist https://gist.github.com/adipola/6a5d7c2f0e13311916c2bddf44a0e13c %}


#### ReduceRows

This functionality uses TensorFlow operations to merge two rows together until there is one row left. It receive the datafram, graph and a ShapeDescription.

User interface:

`def reduceRows(o0: Operation, os: Operation*): Row`

In the next code example. We create a DataFrame with a column named _in_ and two rows. x1 and x2 placeholder for dtype and x- which is an add operation of x1 and x2\. reduceRows, return a Row with value 3 which is the sum of 1.0 and 2.0.

{% gist https://gist.github.com/adipola/c0ae9fc2bf8076f7b240bcafab0472c4 %}


#### ReduceBlocks

Works the same as `ReduceRows` , BUT, it does it on a vector of rows and not row by row.

More used function:

`def reduceBlocks(o0: Operation, os: Operation*): Row`

Code example: Here we create a DataFrame with two columns — _key2_ and _x_. One placeholder names _x1_, one reduce_sum TensorFlow operation named _x_. The reduce functionality return the sum of the rows in the DataFrame according to the desired columns that the reduce_sum named after which is _x_.


{% gist https://gist.github.com/adipola/83134a70435f95437d1edc6aca5d359c %}


<pre name="3290" id="3290" class="graf graf--pre graf-after--p"> -3- Aggregation </pre>

Aggregation is an extra operation for Apache Spark and TensorFlow. It is different from the aggregation functionality in TensorFlow and works with RelationalGroupedDataset. API functionality:

`def aggregate(data: RelationalGroupedDataset, graph: GraphDef, shapeHints: ShapeDescription): DataFrame`

Aggregate receives a RelationalGroupedDataset which is an Apache Spark object, it wraps DataFrame and adds aggregation functionality, a sequence of expressions and a group type.

The aggregate function receives the graph and ShareDescriptiom. It aggregates rows together using reducing transformation on grouped data. This is useful when data is already grouped by key. At the moment, only numerical data is supported.

Code example: In the example, we have a DataFrame with two columns, _key_, and _x_. _x1_ as a placeholder and _x_ as the reduce_sum functionality named _x_.

Using _groupby_ functionality we group the rows by key, and after it, we call _aggregate_ with the operations. We can see in the output that the aggregated was calculated according to the _key_, for the key with value 1- we received 2.1 as the value for _column x_ and for the key with value 2 we received 2.0 as the value for _column x_.

{% gist https://gist.github.com/adipola/36fb197b0ab112d9555542ff57f70c3d %}




## TensorFrames basic process

In all TensorFrames functionality, the DataFrame is sent together with the computations graph. The DataFrame represents the distributed data, meaning in every machine there is a chunck of the data that will go through the graph operations/ transformations.This will happen in every machine with the relevant data. Tungsten binary format is the actual binary in-memory data that goes through the transformation, first to Apache Spark Java object and from there it is sent to TensorFlow Jave API for graph calculations. This all happens in the Spark Worker process, the Spark worker process can spin many tasks which mean various calculation at the same time over the in-memory data. 

![](https://cdn-images-1.medium.com/max/1600/1*59i2Qq2vMjFwr9KYx5E0ig.png)

### Noteworthy

*   DataFrames with scala is **currently** an experimental version.
*   The Scala DSL only features a subset of TensorFlow transforms. 
*   TensorFrames is open source and can be supported [here](https://medium.com/r/?url=https%3A%2F%2Fgithub.com%2Fdatabricks%2Ftensorframes).
*   Python was the first client language supported by TensorFlow and currently supports the most features. More and more of that functionality is being moved into the core of TensorFlow (implemented in C++) and exposed via a [C API](https://medium.com/r/?url=https%3A%2F%2Fwww.tensorflow.org%2Fcode%2Ftensorflow%2Fc%2Fc_api.h). Which later exposed through other languages API, such as Java and JavaScript.
*   Interested in working with Keras? check out [Elephas: Distributed Deep Learning with Keras & Spark](https://medium.com/r/?url=https%3A%2F%2Fgithub.com%2Fmaxpumperla%2Felephas).
*   interested in TensorFrames project on the public cloud? check [this](https://docs.microsoft.com/en-us/learn/paths/azure-fundamentals/?WT.mc_id=devto-blog-adpolak) and [this](https://medium.com/r/?url=https%3A%2F%2Ftowardsdatascience.com%2Fget-started-with-apache-spark-and-tensorflow-on-azure-databricks-163eb3fdb8f3).


**Thank you,**

Follow me on [Twitter](https://medium.com/r/?url=https%3A%2F%2Ftwitter.com%2FAdiPolak), happy to take your suggestions on topics.


**Now that you are an expert on TensorFrames, how will you take it forward?**