Skip to content

[SPARK-16112][SparkR] Programming guide for gapply/gapplyCollect #14090

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
NarineK wants to merge 10 commits into from
Closed

[SPARK-16112][SparkR] Programming guide for gapply/gapplyCollect #14090

Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
29d8a5c
Fixed duplicated documentation problem + separated documentation for …
Jun 20, 2016
698c433
merge with master
Jun 23, 2016
85a4493
Adding programming guide for gapply/gapplyCollect
Jul 7, 2016
7781d1c
removing output format
Jul 7, 2016
c1d7151
Adding R-Spark data-type mapping
Jul 12, 2016
2af7243
bring back div tag got dapplyCollect
Jul 12, 2016
8a2aff3
addressed Felix's comments
Jul 14, 2016
5d34943
merge with master
Jul 14, 2016
19e849f
update data types
Jul 15, 2016
f584416
Fixing a-href and the div for gapply/dapply
Jul 16, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
138 changes: 134 additions & 4 deletions docs/sparkr.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -272,11 +272,11 @@ In SparkR, we support several kinds of User-Defined Functions:

##### dapply
Apply a function to each partition of a `SparkDataFrame`. The function to be applied to each partition of the `SparkDataFrame`
and should have only one parameter, to which a `data.frame` corresponds to each partition will be passed. The output of function
should be a `data.frame`. Schema specifies the row format of the resulting a `SparkDataFrame`. It must match the R function's output.
and should have only one parameter, to which a `data.frame` corresponds to each partition will be passed. The output of function should be a `data.frame`. Schema specifies the row format of the resulting a `SparkDataFrame`. It must match to [data types](#data-type-mapping-between-r-and-spark) of returned value.

<div data-lang="r" markdown="1">
{% highlight r %}

# Convert waiting time from hours to seconds.
# Note that we can apply UDF to DataFrame.
schema <- structType(structField("eruptions", "double"), structField("waiting", "double"),
Expand All @@ -295,8 +295,8 @@ head(collect(df1))

##### dapplyCollect
Like `dapply`, apply a function to each partition of a `SparkDataFrame` and collect the result back. The output of function
should be a `data.frame`. But, Schema is not required to be passed. Note that `dapplyCollect` only can be used if the
output of UDF run on all the partitions can fit in driver memory.
should be a `data.frame`. But, Schema is not required to be passed. Note that `dapplyCollect` can fail if the output of UDF run on all the partition cannot be pulled to the driver and fit in driver memory.

<div data-lang="r" markdown="1">
Copy link
Contributor

@shivaram shivaram Jul 15, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we need a new line before the <div> ? Right now the div markings show up in the generated doc. I've attached a screenshot

screenshot 2016-07-15 14 11 39

{% highlight r %}

Expand All @@ -316,6 +316,136 @@ head(ldf, 3)
{% endhighlight %}
</div>

#### Run a given function on a large dataset grouping by input column(s) and using `gapply` or `gapplyCollect`

##### gapply
Apply a function to each group of a `SparkDataFrame`. The function is to be applied to each group of the `SparkDataFrame` and should have only two parameters: grouping key and R `data.frame` corresponding to
that key. The groups are chosen from `SparkDataFrame`s column(s).
The output of function should be a `data.frame`. Schema specifies the row format of the resulting
`SparkDataFrame`. It must represent R function's output schema on the basis of Spark data types. The column names of the returned `data.frame` are set by user. Below is the data type mapping between R
and Spark.

#### Data type mapping between R and Spark
<table class="table">
<tr><th>R</th><th>Spark</th></tr>
<tr>
<td>byte</td>
<td>byte</td>
</tr>
<tr>
<td>integer</td>
<td>integer</td>
</tr>
<tr>
<td>float</td>
<td>float</td>
</tr>
<tr>
<td>double</td>
<td>double</td>
</tr>
<tr>
<td>numeric</td>
<td>double</td>
</tr>
<tr>
<td>character</td>
<td>string</td>
</tr>
<tr>
<td>string</td>
<td>string</td>
</tr>
<tr>
<td>binary</td>
<td>binary</td>
</tr>
<tr>
<td>raw</td>
<td>binary</td>
</tr>
<tr>
<td>logical</td>
<td>boolean</td>
</tr>
<tr>
<td><a href="https://stat.ethz.ch/R-manual/R-devel/library/base/html/DateTimeClasses.html">POSIXct</a></td>
<td>timestamp</td>
</tr>
<tr>
<td><a href="https://stat.ethz.ch/R-manual/R-devel/library/base/html/DateTimeClasses.html">POSIXlt</a></td>
<td>timestamp</td>
</tr>
<tr>
<td><a href="https://stat.ethz.ch/R-manual/R-devel/library/base/html/Dates.html">Date</a></td>
<td>date</td>
</tr>
<tr>
<td>array</td>
<td>array</td>
</tr>
<tr>
<td>list</td>
<td>array</td>
</tr>
<tr>
<td>env</td>
<td>map</td>
</tr>
</table>

<div data-lang="r" markdown="1">
{% highlight r %}

# Determine six waiting times with the largest eruption time in minutes.
schema <- structType(structField("waiting", "double"), structField("max_eruption", "double"))
result <- gapply(
df,
"waiting",
function(key, x) {
y <- data.frame(key, max(x$eruptions))
},
schema)
head(collect(arrange(result, "max_eruption", decreasing = TRUE)))

## waiting max_eruption
##1 64 5.100
##2 69 5.067
##3 71 5.033
##4 87 5.000
##5 63 4.933
##6 89 4.900
{% endhighlight %}
</div>

##### gapplyCollect
Like `gapply`, applies a function to each partition of a `SparkDataFrame` and collect the result back to R data.frame. The output of the function should be a `data.frame`. But, the schema is not required to be passed. Note that `gapplyCollect` can fail if the output of UDF run on all the partition cannot be pulled to the driver and fit in driver memory.

<div data-lang="r" markdown="1">
{% highlight r %}

# Determine six waiting times with the largest eruption time in minutes.
result <- gapplyCollect(
df,
"waiting",
function(key, x) {
y <- data.frame(key, max(x$eruptions))
colnames(y) <- c("waiting", "max_eruption")
y
})
head(result[order(result$max_eruption, decreasing = TRUE), ])

## waiting max_eruption
##1 64 5.100
##2 69 5.067
##3 71 5.033
##4 87 5.000
##5 63 4.933
##6 89 4.900

{% endhighlight %}
</div>

#### Run local R functions distributed using `spark.lapply`

##### spark.lapply
Expand Down