Как удалить несколько строк в кадре данных Spark в зависимости от позиции (не значения)?

Question

Я хочу выполнить некоторую предварительную обработку данных, используя pyspark, и хочу удалить данные в начале и в конце данных в фрейме данных. Допустим, я хочу удалить первые 30% и последние 30% данных. Я нахожу возможности только на основе значений, использующих where, и нахожу first и last, но не для нескольких. Вот базовый c пример, пока без решения:

import pandas as pd
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("foo").getOrCreate()
cut_factor_start = 0.3 # factor to cut the beginning of the data
cut_factor_stop = 1-cut_factor_start # factor to cut the end of the data
# create pandas dataframe
df = pd.DataFrame({'part':['foo','foo','foo','foo','foo', 'foo'], 'values':[9,1,2,2,6,9]})
# convert to spark dataframe
df = spark.createDataFrame(df)
df.show()

+----+------+
|part|values|
+----+------+
| foo|     9|
| foo|     1|
| foo|     2|
| foo|     2|
| foo|     6|
| foo|     9|
+----+------+

df_length  = df.count()
print('length of df: ' + str(df_length))
cut_start = round(df_length * cut_factor_start)
print('start postion to cut: ' + str(cut_start))
cut_stop  = round(df_length * (cut_factor_stop))
print('stop  postion to cut: ' + str(cut_stop))

length of df: 6
start postion to cut: 2
stop  postion to cut: 4

На чем я хочу это основано расчеты:

+----+------+
|part|values|
+----+------+
| foo|     1|
| foo|     2|
| foo|     2|
+----+------+

Answer 1

Другой способ - использовать between после присвоения номера строки:

import pyspark.sql.functions as F
from pyspark.sql import Window

rnum= F.row_number().over(Window.orderBy(F.lit(0)))
output = (df.withColumn('Rnum',rnum)
        .filter(F.col("Rnum").between(cut_start, cut_stop)).drop('Rnum'))

---

output.show()

+----+------+
|part|values|
+----+------+
| foo|     1|
| foo|     2|
| foo|     2|
+----+------+

Answer 2

В Scala, можно добавить уникальный столбец «id», а затем функции «limit» и «кроме»:

val dfWithIds = df.withColumn("uniqueId", monotonically_increasing_id())
dfWithIds
  .limit(stopPostionToCut)
  .except(dfWithIds.limit(startPostionToCut - 1))
  .drop("uniqueId")