Spark-Scala: Как сравнить даты в строках различной длины и вернуть минимальное значение?

#scala #apache-spark #datetime

Вопрос:

У меня есть набор данных следующего типа, где первый элемент представляет userid , а строка второго элемента и далее представляет собой value для этого userid :

 1004,bb5469c5|2021-09-19 01:25:30,4f0d-bb6f-43cf552b9bc6|2021-09-25 05:12:32,1954f0f|2021-09-19 01:27:45,4395766ae|2021-09-19 01:29:13,
1018,36ba7a7|2021-09-19 01:33:00,
1020,23fe40-4796-ad3d-6d5499b|2021-09-19 01:38:59,77a90a1c97b|2021-09-19 01:34:53,
1022,3623fe40|2021-09-19 01:33:00,
1028,6c77d26c-6fb86|2021-09-19 01:50:50,f0ac93b3df|2021-09-19 01:51:11,
1032,ac55-4be82f28d|2021-09-19 01:54:20,82229689e9da|2021-09-23 01:19:47,

Я должен проанализировать value часть каждой строки и вернуть из них минимальную дату. Строки имеют переменную/динамическую длину, и нет установленного ограничения на длину строки.

Я попытался передать их в функцию:

 val strFun = (str: String) =>{
    if (str != null) {
      val str_split = str.replaceAll("""|""",",").split(",")
      }
    }
    (str_split(0),str_split(1))
  };

Но ограничение, на котором я застрял, состоит в том, чтобы определить количество дат, которые должны быть возвращены для сравнения, и выбрать из них минимальную дату. Как это можно сделать?

1. В течение двух дней вы продолжаете принимать и не принимать мой ответ (4 раза, если я не ошибаюсь). Может быть, вы могли бы сами попытаться перенять текущие идеи в последнюю версию вашего вопроса? Пожалуйста, помните, что SO не является бесплатной услугой кодирования.

2. Я понимаю, что SO не является бесплатной службой кодирования. Я сам пытался реализовать решение в коде и использую опцию «Принято» и «не принято» в качестве закладки для своего подхода. Я также понимаю, что это несправедливо по отношению к вам, и исправляю это.

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

4. Не волнуйся! Все в порядке. Я был просто удивлен, что ты так часто меняла свое мнение 🙂

Ответ №1:

Сначала разделите строку. В результате получается строковый массив. Использование transform из каждого элемента массива извлекается (используется regexp_extract ) строка даты и преобразуется в метку времени. Теперь у нас есть массив временных меток, из которых мы берем минимальное значение.

 import org.apache.spark.sql.functions._
df.withColumn("min_date", split('column, ","))
  .withColumn("min_date", array_min(transform('min_date,
      c => to_timestamp(regexp_extract(c, "\|(.*)$", 1)))))
  .show()

Выход:

  ------ -------------------- ------------------- 
|userid|              column|           min_date|
 ------ -------------------- ------------------- 
|  1004|value=bb5469c5|20...|2021-09-19 01:25:30|
| 1018.|value=36ba7a7|202...|2021-09-19 01:33:00|
|  1020|value=23fe40-4796...|2021-09-19 01:34:53|
| 1022.|value=3623fe40|20...|2021-09-19 01:33:00|
|  1028|value=6c77d26c-6f...|2021-09-19 01:50:50|
|  1032|value=ac55-4be82f...|2021-09-19 01:54:20|
 ------ -------------------- -------------------

1. Версия 3 против версии 2.4?

2. Да, я тоже проверил это. Это вызвало у меня проблему, потому что запущенная версия Spark составляет 2.4

3. Хороший результат 9999

4. @PixieDev не могли бы вы попробовать заменить вторую withColumn строку на .withColumn("min_date", array_min(expr("transform(min_date, c -> to_timestamp(regexp_extract(c, '\\|(.*)$', 1)))"))) ? Я думаю, что это должно сработать для 2.4

5. @вернер, Это сработало, спасибо!