Как получить только одну строку, которая чуть меньше указанного значения в кадре данных искры

Question

У меня есть датафрейм, как показано ниже -

Id,timestamp
100,1
200,2
300,3
400,4
500,5
600,6

А теперь я хочу получить только одну строку, значение которой чуть меньше отметки времени 5. Эквивалентный SQL-запрос: select * from table1 where id <5 order by timestamp desc limit 1.

Но в искровом запросе я не хочу выполнять упорядочение по операции, потому что размер наших данных слишком велик, поэтому это займет слишком много времени. Возможно ли в spark сделать то же самое, не используя order by?

Answer 1

Хотя остальные ответы верны (вы можете использовать свой запрос непосредственно во временном представлении или orderBy + limit), похоже, ни один из них не решает вашу главную проблему:

Но в запросе Spark я не хочу выполнять упорядочение по операции, потому что размер наших данных слишком велик, поэтому это займет слишком много времени. Возможно ли в Spark сделать то же самое, не используя order by?

Давайте посмотрим, что происходит под одеялом:

df = spark.createDataFrame(
    [(100, 1), (200, 2), (300, 3), (400, 4), (500, 5), (600, 6)],
    ("id", "timestamp"))

df.filter("timestamp < 5").orderBy(df["timestamp"].desc()).limit(1).explain()
# == Physical Plan ==
# TakeOrderedAndProject(limit=1, orderBy=[timestamp#35L DESC NULLS LAST], output=[id#34L,timestamp#35L])
# +- *(1) Filter (isnotnull(timestamp#35L) && (timestamp#35L < 5))
#    +- Scan ExistingRDD[id#34L,timestamp#35L]

Как видите, сортировки вообще нет. Спарк проанализировал план, вместо полной сортировки применил TakeOrderedAndProject. В результате он будет занимать только первое место в каждом разделе.

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

Вывод? Пока предельное значение невелико, здесь не о чем беспокоиться (при больших лимитах все может стать уродливым).

Answer 2

Если вы хотите избежать ordering из-за большого размера кадра данных, рассмотрите возможность использования reduce, чтобы получить максимум только после фильтрации:

val df = Seq(
  (100, 1),
  (200, 2),
  (300, 3),
  (400, 4),
  (500, 5),
  (600, 6)
).toDF("id", "ts")

df.where($"ts" < 5).reduce( (acc, r) =>
  if (r.getAs[Int]("ts") > acc.getAs[Int]("ts")) r else acc
)
// res1: org.apache.spark.sql.Row = [400,4]  

Answer 3

Прежде всего, вы можете запустить этот SQL-запрос прямо в spark:

dataframe.createOrReplaceTempView("table1")

spark.sql("select * from table1 where id <5 order by timestamp desc limit 1")

Кроме того, вы знаете, что id (я вижу, вы имели в виду timestamp) ровно 4:

dataframe.where("timestamp = 4")

Answer 4

Другой способ:

import pyspark.sql.functions as f
df.filter("id<5").orderBy(f.col("timestamp").desc()).first()

Answer 5

Короче: сортировка, фильтрация по значению <указано, выберите последний элемент результата. </p>