%md # Higher-Order and Lambda Functions: Explore Complex and Structured Data in SQL 

%md This tutorial walks you through four higher-order functions. While this in-depth [blog](https://databricks.com/blog/2017/05/24/working-with-nested-data-using-higher-order-functions-in-sql-on-databricks.html) explains the concepts, justifications, and motivations of _why_ handling complex data types such as arrays are important in SQL, and equally explains why their existing implementation are inefficient and cumbersome, this tutorial shows _how_ to use higher-order functions in SQL in processing structured data and arrays in IoT device events. In particular, they come handy and you can put them to good use if you enjoy functional programming and can quickly and can efficiently write a lambda expression as part of these higher-order SQL functions. 

This tutorial explores four functions and how you can put them to a wide range of uses in your processing and transforming array types:

* `transform()`

* `filter()`

* `exists()`

* `aggregate()`

The takeaway from this short tutorial is that there exists myriad ways to slice and dice nested JSON structures with Spark SQL utility functions. These dedicated higher-order functions are primarily suited to manipulating arrays in Spark SQL, making it easier and the code more concise when processing table values with arrays or nested arrays. 

%md Let's create a simple JSON schema with attributes and values, with at least two attributes that are arrays, namely _temp_ and _c02_level_

%md This helper Python function converts a JSON string into a [Python DataFrame](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrame.html#pyspark-sql-dataframe).

%md Using the schema above, create a complex JSON stucture and create into a Python DataFrame. Display the DataFrame gives us two columns: a key (dc_id) and value (source), which is JSON string with embedded nested structure.

%md By examining its schema, you can notice that the DataFrame schema reflects the above defined schema, where two of its elments are are arrays of integers. 

%md Employ `explode()` to explode the column `source` into its individual columns.

%md Now you can work with the `value` column, which a is `struct`, to extract individual fields by using their names.

%md For sanity let's ensure what was created as DataFrame was preserved and adherent to the schema declared above while exploding and extracting individual data items.

%md Now, since this tutorial is less about DataFrames API and more about higher-order functions and lambdas in SQL, create a temporary table or view and start using the higher-order SQL functions mentioned above.

%md The table was created as columns in your DataFrames and it reflects its schema.

%md ## SQL Higher-Order Functions and Lambda Expressions

%md ### How to use `transform()`

%md Its functional signature, `transform(values, value -> lambda expression)`, has two components:

1. `transform(values..)` is the higher-order function. This takes an array and an anonymous function as its input. Internally, `transform` takes care of setting up a new array, applying the anonymous function to each element, and then assigning the result to the output array.

2.  The `value -> expression`  is an anonymous function. The function is further divided into two components separated by a `->` symbol:

  * **The argument list**: This case has only one argument: value. You can specify multiple arguments by creating a comma-separated list of arguments enclosed by parenthesis, for example: `(x, y) -> x + y.`

  * **The body**: This is a SQL expression that can use the arguments and outer variables to calculate the new value. 

In short, the programmatic signature for `transform()` is as follows:

`transform(array<T>, function<T, U>): array<U>`

This produces an array by applying a function<T, U> to each element of an input array<T>.

Note that the functional programming equivalent operation is `map`. This has been named transform in order to prevent confusion with the map expression (that creates a map from a key value expression).

This basic scheme for `transform(...)` works the same way as with other higher-order functions, as you will see shortly.

%md 

The following query transforms the values in an array by converting each elmement's temperature reading from Celsius to Fahrenheit.

Let's transform (and hence convert) all our Celsius reading into Fahrenheit. (Use conversion formula: `((C * 9) / 5) + 32`) The lambda expression here is the formula to convert **C->F**.

Now, `temp` and `((t * 9) div 5) + 32` are the arguments to the higher-order function `transform()`. The anonymous function will iterate through each element in the array, `temp`, apply the function to it, and transforming its value and placing into an output array. The result is a new column with tranformed values: `fahrenheit_temp`.

%md While the above example generates transformed values, this example uses a Boolean expression as a lambda function and generates a boolean array of results instead of values, since the expression 

`t->t > 1300` is a predicate, resulting into a true or false.

%md ### How to use `filter()`

%md As with `transform`, `filter` has a similar signature, `filter(array<T>, function<T, Boolean>): array<T>`

Unlike `transform()` with a boolean expression, it produces an output array from an input array by *only* adding elements for which predicate `function<T, Boolean>` holds.

For instance, let's include only readings in our `c02_levels` that exceed dangerous levels (`cO2_level > 1300`). Again the functional signature is not dissimilar to `transform()`. However, note the difference in how `filter()` generated the resulting array compared to _transform()_ with similar lambda expression.

%md Notice when the lambda's predicate expression is reversed, the resulting array is empty. That is, it does not evaluate to values true or false as it did in `tranform()`.

%md ### How to use `exists()`

%md A mildly different functional signature than the above two functions, the idea is simple and same: 

`exists(array<T>, function<T, V, Boolean>): Boolean`

Return true if predicate `function<T, Boolean>` holds for any element in input array.

In this case you can iterate through the `temp` array and see if a particular value exists in the array. Let's acertain if any of your values contains 45 degrees Celsius or determine of a c02 level in any of the readings equals to 1570.

%md ### How to use `reduce()`

%md By far this function and its method is more advanced than others. It also allows you to do aggregation, as seen in the next section.

Its signature allows us to some extra bit with the last lambda expression as its functional argument.

`reduce(array<T>, B, function<B, T, B>, function<B, R>): R`

Reduce the elements of `array<T>` into a single value `R` by merging the elements into a buffer B using `function<B, T, B>` and by applying a finish `function<B, R>` on the final buffer. The initial value `B` is determined by a `zero` expression. 

The finalize function is optional, if you do not specify the function the finalize function the identity function (`id -> id`) is used.

This is the only higher-order function that takes two lambda functions.

For instance, if you want to compute an average of the temperature readings, use lambda expressions: The first one accumulates all the results into an internal temporary buffer, and the second function applies to the final accumulated buffer. With respect to our signature above, `B` is `0`; `function<B,T,B>` is `t + acc`, and `function<B,R>` is `acc div size(temp)`. Furthermore, in the finalize lambda expression, convert the average temperature to Fahrenheit.