Spark groupby, сортировка значений, затем взять первый и последний

Question

Я использую Apache Spark и у меня есть фрейм данных, который выглядит следующим образом:

scala> df.printSchema
root
 |-- id: string (nullable = true)
 |-- epoch: long (nullable = true)


scala> df.show(10)
+--------------------+-------------+
|                 id |        epoch|
+--------------------+-------------+
|6825a28d-abe5-4b9...|1533926790847|
|6825a28d-abe5-4b9...|1533926790847|
|6825a28d-abe5-4b9...|1533180241049|
|6825a28d-abe5-4b9...|1533926790847|
|6825a28d-abe5-4b9...|1532977853736|
|6825a28d-abe5-4b9...|1532531733106|
|1eb5f3a4-a68c-4af...|1535383198000|
|1eb5f3a4-a68c-4af...|1535129922000|
|1eb5f3a4-a68c-4af...|1534876240000|
|1eb5f3a4-a68c-4af...|1533840537000|
+--------------------+-------------+
only showing top 10 rows

Я хочу сгруппировать по полю id, чтобы собрать все метки времени эпохи вместе для id.Затем я хочу отсортировать эпохи по возрастанию, а затем взять первую и последнюю эпохи.

Я использовал следующий запрос, но значения эпох first и last отображаются в том порядке, в которомони появляются в исходном фрейме данных.Я хочу, чтобы первое и последнее извлекались из отсортированного по возрастанию порядка.

scala> val df2 = df2.groupBy("id").
                 agg(first("epoch").as("first"), last("epoch").as("last"))

scala> df2.show()
+--------------------+-------------+-------------+                              
|                  id|        first|         last|
+--------------------+-------------+-------------+
|4f433f46-37e8-412...|1535342400000|1531281600000|
|d0cba2f9-cc04-42c...|1535537741000|1530448494000|
|6825a28d-abe5-4b9...|1533926790847|1532531733106|
|e963f265-809c-425...|1534996800000|1534996800000|
|1eb5f3a4-a68c-4af...|1535383198000|1530985221000|
|2e65a033-85ed-4e4...|1535660873000|1530494913413|
|90b94bb0-740c-42c...|1533960000000|1531108800000|
+--------------------+-------------+-------------+

Как получить первое и последнее из списка эпох, отсортированного по возрастанию?

Answer 1

Вы можете просто использовать min и max и приводить полученные столбцы к строке.Вот один из способов сделать это

   import org.apache.spark.sql.functions._
val df = Seq(("6825a28d-abe5-4b9",1533926790847.0),
("6825a28d-abe5-4b9",1533926790847.0),
("6825a28d-abe5-4b9",1533180241049.0),
("6825a28d-abe5-4b9",1533926790847.0),
("6825a28d-abe5-4b9",1532977853736.0),
("6825a28d-abe5-4b9",1532531733106.0),
("1eb5f3a4-a68c-4af",1535383198000.0),
("1eb5f3a4-a68c-4af",1535129922000.0),
("1eb5f3a4-a68c-4af",1534876240000.0),
("1eb5f3a4-a68c-4af",1533840537000.0)).toDF("id","epoch").withColumn("epoch",($"epoch"/1000.0).cast("timestamp"))

    +-----------------+--------------------+
|               id|               epoch|
+-----------------+--------------------+
|6825a28d-abe5-4b9|2018-08-10 18:46:...|
|6825a28d-abe5-4b9|2018-08-10 18:46:...|
|6825a28d-abe5-4b9|2018-08-02 03:24:...|
|6825a28d-abe5-4b9|2018-08-10 18:46:...|
|6825a28d-abe5-4b9|2018-07-30 19:10:...|
|6825a28d-abe5-4b9|2018-07-25 15:15:...|
|1eb5f3a4-a68c-4af| 2018-08-27 15:19:58|
|1eb5f3a4-a68c-4af| 2018-08-24 16:58:42|
|1eb5f3a4-a68c-4af| 2018-08-21 18:30:40|
|1eb5f3a4-a68c-4af| 2018-08-09 18:48:57|
+-----------------+--------------------+

    val df1 = df.groupBy("id").agg(min($"epoch").cast("string").as("first"), max($"epoch").cast("string"). as("last"))
df1.show

    +-----------------+--------------------+--------------------+
|               id|               first|                last|
+-----------------+--------------------+--------------------+
|6825a28d-abe5-4b9|2018-07-25 15:15:...|2018-08-10 18:46:...|
|1eb5f3a4-a68c-4af| 2018-08-09 18:48:57| 2018-08-27 15:19:58|
+-----------------+--------------------+--------------------+


    df1: org.apache.spark.sql.DataFrame = [id: string, first: string ... 1 more field]

Answer 2

Функции

first и last не имеют смысла при применении вне контекста Window.Принятое значение является чисто произвольным.

Вместо этого следует

• Использовать функции min / max, если логика соответствует основным правилам упорядочения (буквенно-цифровая длястроки, массивы и структуры, числовые для чисел).

• Набор данных строгого типа с map -> groupByKey -> reduceGroups или groupByKey -> mapGroupsв противном случае.