Как взять подмножество записей за временное окно? - PullRequest
Новая
       19

Как взять подмножество записей за временное окно?

2 голосов
/ 21 октября 2019

У меня очень большой фрейм данных (450000 строк) с данными датчика и отметкой времени, что-то вроде этого: 

+--------+-----------+-----------+------------+-----------+
|Time [s]|   Sensor1 |   Sensor2 |    Sensor3 |   Sensor4 | 
+--------+-----------+-----------+------------+-----------+
| 0.00198|-0.55999756|-0.19271851|   1.1320496|   1.373291| 
| 0.00398| -1.2171936|  1.0081482|  0.25726318| 0.61035156| 
| 0.00598|-0.29586792|  1.4437866|  -1.1341858|   1.373291| 
| 0.00798|  1.4489746| 0.39047241|  -1.4620972|-0.30517578| 
| 0.00998|  1.5341187| -1.1869812| -0.19256592|-0.15258789| 
| 0.01198| 0.04196167| -1.2620544|   1.1372375| 0.45776367| 
| 0.01398| -1.0899353| 0.19500732|  0.79772949|  1.8310547| 
| 0.01598| -0.6300354| 0.77346802| -0.69030762| 0.61035156| 
| 0.01798| 0.95153809| 0.40786743| -0.96694946|        0.0| 
| 0.01998|  1.5705872|-0.75668335| 0.063323975| 0.91552734|
| 0.02198| 0.29678345| -1.4421082|   1.1439514| -1.0681152|
| 0.02398| -1.3595581|-0.25726318|   1.4170837| 0.45776367|
+--------+-----------+-----------+------------+-----------+

Мне нужно сгладить эти данные в окне времени и добавить в список.

Например, если окно составляет 10 мс, я бы взял 5 данных каждого датчика сверху и добавил к списку, который выглядел бы так: 

[[-0.55999756, -0.19271851, 1.1320496, 1.373291, -1.2171936, 1.0081482, 0.25726318, 0.61035156, -0.29586792, 1.4437866, -1.1341858, 1.373291, 1.4489746, 0.39047241, -1.4620972, -0.30517578, 1.5341187, -1.1869812, -0.19256592, -0.15258789]
... ]

Яв настоящее время это достигается с помощью следующего кода: 

mylist=[]
df= df.withColumn("row", row_number().over(Window.orderBy(monotonically_increasing_id())))
for m in range(n+1, df.count()+n+1, n):
  newdf= df.filter((col("row")>(m-n)) & (col("row")<m))
  flatlist= newdf.select("Sensor1", "Sensor2", "Sensor3", "Sensor4").rdd.flatMap(lambda x: x).collect()
  mylist.append(flatlist)

, где m и n - границы моего окна.

Это работает, но для большого окна и большого кадра данных это занимает вечность (возможно, потому чтособирать ()?). Есть ли более эффективный способ получить тот же результат?

С помощью панд я могу сделать следующее, но является ли он более эффективным? (Я бы предпочел сделать все это с помощью Spark для распараллеливания) 

pandasdf = df.toPandas()
flatlist=[pandasdf.values.flatten().tolist()]

Ответы [ 2 ]

0 голосов
/ 22 октября 2019

Это вернет данные через каждые 10 мсек. Однако это будет массив массивов. Не уверен, если вам это нужно в одном массиве. 

df = df.withColumn('sensorDataArr', F.array('Sensor1', 'Sensor2', 'Sensor3', 'Sensor4'))
df = df.withColumn('grpNum', F.floor(df.Time * 100))
df_g = df.groupBy('grpNum').agg(F.collect_list('sensorDataArr').alias('sensorData'))
0 голосов
/ 22 октября 2019

tl; dr Использование оператора groupBy (возможно, со window стандартной функцией), за которым следует collect_list стандартная функция.

Возможно, вы захотите использовать пользовательскую функцию(UDF), чтобы взять только первые 5 элементов из collect_list.

Я не использую Python / pyspark, поэтому могу помочь больше.

...