Как правильно фильтровать данные периода в Spark?

Question

Я новичок в Spark и мне нужен совет.

У меня есть паркет с большим количеством колонн.Я хочу отфильтровать данные периода для конкретных номеров мобильных телефонов.

Имеются следующие входные данные:

val dateFrom = "2018-10-01"
val dateTo = "2018-11-05"
val mobile_numbers = "7778529636,745128598,7777533575"

Паркет имеет такую ​​структуру:

| START_DATE          | MOBILE_NUMBER | STATUS | OTHER_COLUMNS
|---------------------|---------------|--------|--------------
| 2018-11-28 20:00:00 | 7778541536    | IN     | ************
| 2018-11-29 00:00:00 | 7786984525    | OUT    | ************
| 2018-11-30 09:00:00 | 7986984525    | IN     | ************
| 2018-12-01 00:30:00 | 7685984425    | OUT    | ************
| 2018-12-02 12:00:00 | 7586984525    | IN     | ************

КОД :

spark.read.parquet("fs://path/file.parquet").filter(???)

Answer 1

Я предполагаю, что вы хотите выбрать те строки, которые попадают в определенный диапазон дат и имеют значения столбца MOBILE_NUMBER среди списка определенных номеров.

Сначала вы должны убедиться, что ваши данные в вашем ScalaКод и внутри фрейма данных находятся в одном формате, чтобы вы могли сравнить их правильно.Например:

val dateFrom = "2018-10-01 00:00:00"
val dateTo = "2018-11-05 23:59:59"
val mobileNumbers = Vector("7778529636", "745128598", "7777533575")

Обратите внимание, что я представляю номера мобильных телефонов в виде строк, поскольку вы можете получить неожиданные результаты для номеров с нулями в начале.Даты также являются строками.

Тогда запрос фильтра может выглядеть следующим образом:

import spark.implicits._  // for the $"something" syntax

spark.read.parquet("...")
  .filter(
    $"START_DATE".between(dateFrom, dateTo) && $"MOBILE_NUMBER".isin(mobileNumbers: _*)
  )

В идеале даты должны быть в формате, который не допускает двусмысленности, например, числовая метка времени,потому что в противном случае недопустимые значения, такие как 2018-10-01 56:78:90, могут нарушить сравнение, но в большинстве случаев это должно сработать, и, если это действительно необходимо, вы можете написать UDF, используя метод org.apache.spark.sql.functions.udf, чтобы убедиться, что оно полностью корректно.