Skip to content

Commit 5982162

Browse files
yoda-monHyukjinKwon
authored andcommitted
[SPARK-36976][R] Add max_by/min_by API to SparkR
### What changes were proposed in this pull request? Add max_by/min_by to SparkR ### Why are the changes needed? for sparkr users' convenience ### Does this PR introduce _any_ user-facing change? yes, new methods are added ### How was this patch tested? unit test Closes #34258 from yoda-mon/max-by-min-by-r. Authored-by: Leona Yoda <yodal@oss.nttdata.com> Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>
1 parent bc7e4f5 commit 5982162

4 files changed

Lines changed: 72 additions & 0 deletions

File tree

R/pkg/NAMESPACE

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -352,9 +352,11 @@ exportMethods("%<=>%",
352352
"map_values",
353353
"map_zip_with",
354354
"max",
355+
"max_by",
355356
"md5",
356357
"mean",
357358
"min",
359+
"min_by",
358360
"minute",
359361
"monotonically_increasing_id",
360362
"month",

R/pkg/R/functions.R

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1479,6 +1479,29 @@ setMethod("max",
14791479
column(jc)
14801480
})
14811481

1482+
#' @details
1483+
#' \code{max_by}: Returns the value associated with the maximum value of ord.
1484+
#'
1485+
#' @rdname column_aggregate_functions
1486+
#' @aliases max_by max_by,Column-method
1487+
#' @note max_by since 3.3.0
1488+
#' @examples
1489+
#'
1490+
#' \dontrun{
1491+
#' df <- createDataFrame(
1492+
#' list(list("Java", 2012, 20000), list("dotNET", 2012, 5000),
1493+
#' list("dotNET", 2013, 48000), list("Java", 2013, 30000)),
1494+
#' list("course", "year", "earnings")
1495+
#' )
1496+
#' tmp <- agg(groupBy(df, df$"course"), "max_by" = max_by(df$"year", df$"earnings"))
1497+
#' head(tmp)}
1498+
setMethod("max_by",
1499+
signature(x = "Column", y = "Column"),
1500+
function(x, y) {
1501+
jc <- callJStatic("org.apache.spark.sql.functions", "max_by", x@jc, y@jc)
1502+
column(jc)
1503+
})
1504+
14821505
#' @details
14831506
#' \code{md5}: Calculates the MD5 digest of a binary column and returns the value
14841507
#' as a 32 character hex string.
@@ -1531,6 +1554,29 @@ setMethod("min",
15311554
column(jc)
15321555
})
15331556

1557+
#' @details
1558+
#' \code{min_by}: Returns the value associated with the minimum value of ord.
1559+
#'
1560+
#' @rdname column_aggregate_functions
1561+
#' @aliases min_by min_by,Column-method
1562+
#' @note min_by since 3.3.0
1563+
#' @examples
1564+
#'
1565+
#' \dontrun{
1566+
#' df <- createDataFrame(
1567+
#' list(list("Java", 2012, 20000), list("dotNET", 2012, 5000),
1568+
#' list("dotNET", 2013, 48000), list("Java", 2013, 30000)),
1569+
#' list("course", "year", "earnings")
1570+
#' )
1571+
#' tmp <- agg(groupBy(df, df$"course"), "min_by" = min_by(df$"year", df$"earnings"))
1572+
#' head(tmp)}
1573+
setMethod("min_by",
1574+
signature(x = "Column", y = "Column"),
1575+
function(x, y) {
1576+
jc <- callJStatic("org.apache.spark.sql.functions", "min_by", x@jc, y@jc)
1577+
column(jc)
1578+
})
1579+
15341580
#' @details
15351581
#' \code{minute}: Extracts the minute as an integer from a given date/timestamp/string.
15361582
#'

R/pkg/R/generics.R

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1190,10 +1190,18 @@ setGeneric("map_values", function(x) { standardGeneric("map_values") })
11901190
#' @name NULL
11911191
setGeneric("map_zip_with", function(x, y, f) { standardGeneric("map_zip_with") })
11921192

1193+
#' @rdname column_aggregate_functions
1194+
#' @name NULL
1195+
setGeneric("max_by", function(x, y) { standardGeneric("max_by") })
1196+
11931197
#' @rdname column_misc_functions
11941198
#' @name NULL
11951199
setGeneric("md5", function(x) { standardGeneric("md5") })
11961200

1201+
#' @rdname column_aggregate_functions
1202+
#' @name NULL
1203+
setGeneric("min_by", function(x, y) { standardGeneric("min_by") })
1204+
11971205
#' @rdname column_datetime_functions
11981206
#' @name NULL
11991207
setGeneric("minute", function(x) { standardGeneric("minute") })

R/pkg/tests/fulltests/test_sparkSQL.R

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2292,6 +2292,22 @@ test_that("group by, agg functions", {
22922292
unlink(jsonPath3)
22932293
})
22942294

2295+
test_that("SPARK-36976: Add max_by/min_by API to SparkR", {
2296+
df <- createDataFrame(
2297+
list(list("Java", 2012, 20000), list("dotNET", 2012, 5000),
2298+
list("dotNET", 2013, 48000), list("Java", 2013, 30000))
2299+
)
2300+
gd <- groupBy(df, df$"_1")
2301+
2302+
actual1 <- agg(gd, "_2" = max_by(df$"_2", df$"_3"))
2303+
expect1 <- createDataFrame(list(list("dotNET", 2013), list("Java", 2013)))
2304+
expect_equal(collect(actual1), collect(expect1))
2305+
2306+
actual2 <- agg(gd, "_2" = min_by(df$"_2", df$"_3"))
2307+
expect2 <- createDataFrame(list(list("dotNET", 2012), list("Java", 2012)))
2308+
expect_equal(collect(actual2), collect(expect2))
2309+
})
2310+
22952311
test_that("pivot GroupedData column", {
22962312
df <- createDataFrame(data.frame(
22972313
earnings = c(10000, 10000, 11000, 15000, 12000, 20000, 21000, 22000),

0 commit comments

Comments
 (0)