How to take a subset of records per time window?

user_3798240 0
Viewed 64

I have a very large dataframe (450000 rows) with sensor data and timestamp, something like this:

+--------+-----------+-----------+------------+-----------+
|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|
+--------+-----------+-----------+------------+-----------+

I need to flatten this data over a time window and append to a list.

For example, if the window is 10ms, then I would take 5 of each sensor data from above and add to the list, which would look like this:

[[-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]
... ]

I'm currently achieving this with the following code:

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)

where m and n are the boundaries of my window.

This works but for a large window and large dataframe it takes forever (maybe because of collect()?). Is there a more efficient way to get the same result?

With Pandas I can do the following, but is it more efficient? (I would rather do it all with Spark for parallelization)

pandasdf = df.toPandas()
flatlist=[pandasdf.values.flatten().tolist()]
pyspark-sql apache-spark
2 Answers

tl;dr Use groupBy operator (perhaps with window standard function) followed by collect_list standard function.

You may want to use a user-defined function (UDF) to take just the very first 5 elements out of the collect_list.

I'm not using Python / pyspark so can help more.

user_13053440

This would return the data in every 10 ms window. However it will be array of arrays. Not sure if you need it in single array.

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'))
user_100638280
Related