как я могу отбросить дубликаты на ["x", "y"] без перетаскивания искрового фрейма данных, уже разделенного на "x"

Question

У меня есть искровой фрейм данных, который уже был перераспределен по столбцу x:

df2 = df1.reparition("x")

Я хотел бы отбрасывать дубликаты на x и другой столбец без тасования, так как тасование очень длинное в этом конкретном случае. case.

df3 = df2.dropDuplicates(subset=["x","y"])

edit: Очевидно, что существующая реализация dropDuplicates не поддерживает не тасование. Есть ли способ добиться аналогичных результатов, используя sql оконные функции над y, предполагая, что данные были недавно разделены на x.

Answer 1

Я думаю, что произойдет случайное перемешивание, но только для выполнения repartition("x"). Последующие dropDuplicates() будут затем сортировать разделы по key=["x","y"], а затем агрегировать, чтобы получить first строку для каждого ключа. Поскольку все строки для x уже находятся в одном разделе, дополнительная перестановка не потребуется.

ОБНОВЛЕНИЕ

Позволяет выполнить быстрый тест:

[user@gateway ~]# pyspark
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.0-cdh6.x-SNAPSHOT
      /_/

Using Python version 2.7.5 (default, Apr  9 2019 14:30:50)
SparkSession available as 'spark'.

>>> df1 = spark.createDataFrame([{'x': 1, 'y': 1, 'z': 1},{'x': 1, 'y': 1, 'z': 2},{'x': 1, 'y': 2, 'z': 2},{'x': 2, 'y': 1, 'z': 1}])
>>> df1.printSchema()
root
 |-- x: long (nullable = true)
 |-- y: long (nullable = true)
 |-- z: long (nullable = true)

>>> df2 = df1.repartition("x")
>>> df3 = df2.dropDuplicates(subset=["x","y"])
>>> df3.explain()
== Physical Plan ==
*(1) HashAggregate(keys=[x#0L, y#1L], functions=[first(z#2L, false)])
+- *(1) HashAggregate(keys=[x#0L, y#1L], functions=[partial_first(z#2L, false)])
   +- Exchange hashpartitioning(x#0L, 200)
      +- Scan ExistingRDD[x#0L,y#1L,z#2L]

>>> df3.show()
+---+---+---+
|  x|  y|  z|
+---+---+---+
|  1|  2|  2|
|  1|  1|  1|
|  2|  1|  1|
+---+---+---+

>>>

На плане показан один оператор Exchange (перемешивание на x), затем HashAggregate s, работающие с данным разделом, чтобы получить partial_first для каждой пары x,y и взять первую строку (даже без сортировки).