如果在列表中的任何一组点之间过滤Spark DataFrame

我有一个有timestamp，asset（一个字符串），tag（一个字符串）和value（一个double）列的spark数据框（使用scala接口）。下面是它的一个摘录：如果在列表中的任何一组点之间过滤Spark DataFrame

+--------------------+-----+--------+-------------------+ 
|   timestamp|asset|  tag|    value| 
+--------------------+-----+--------+-------------------+ 
|2013-01-03 23:36:...| G4| BTGJ2_2|  116.985626221| 
|2013-01-15 00:36:...| G4| TTXD1_6|  66.887382507| 
|2013-01-05 13:03:...| G4|TTXD1_22|  40.913497925| 
|2013-01-12 04:43:...| G4|TTXD1_23|  60.834510803| 
|2013-01-08 17:54:...| G4| LTB1D|  106.534744263| 
|2013-01-02 04:15:...| G4| WEXH|  255.981292725| 
|2013-01-07 10:54:...| G4| BTTA1_7|  100.743843079| 
|2013-01-05 11:29:...| G4| CDFH_10|  388.560668945| 
|2013-01-10 09:10:...| G4| LTB1D|  112.226242065| 
|2013-01-13 15:09:...| G4|TTXD1_15|  63.970848083| 
|2013-01-15 01:23:...| G4| TTIB|  67.993904114|

我也有一个Array[List[Timestamp]]，其中每个List为大小两种，并拥有启动和感兴趣的间隔结束时间。例如：

event_times: Array[List[java.sql.Timestamp]] = Array(List(2013-01-02 00:00:00.0, 2013-01-02 12:00:00.0), List(2013-01-10 00:00:00.0, 2013-01-12 06:00:00.0))

持有的兴趣两个区间：一个是从午夜至12:00 2013年1月2日，另一个是午夜2013年1月10日至6:00 2013年1月12日

这是我的问题：我如何过滤数据帧以返回值，使得时间戳记在间隔的任何？对于任何一个时间间隔，我可以做

df.filter(df("timestamp").between(start, end))

因为我不知道有多少元素在Array（多少间隔我），我不能只是有很长的一系列过滤器。

对于上面的例子，我想保持行4,6和9

我现在是在一个Array循环，我越来越为每一个相应的子集。但是，这可能比将它放在一个很大的过滤器中要慢一些？

来源

2016-08-04 kgully

对不起，我没有得到你想要的。你能举一个明确的例子吗？ –

@ThiagoBaldim我刚刚更新了一个例子。 – kgully

您可以将时间戳列表转换为DataFrame，并将其与相应时间戳中的初始DataFrame结合使用。我创建了一个简单的例子来说明这个过程：

//Dummy data 
val data = List(
    ("2013-01-02 00:30:00.0", "116.985626221"), 
    ("2013-01-03 00:30:00.0", "66.887382507"), 
    ("2013-01-11 00:30:00.0", "12.3456") 
) 

//Convert data to DataFrame 
val dfData = sc.parallelize(data).toDF("timestamp", "value") 

//Timestamp intervals list 
val filterList = Array(
    List("2013-01-02 00:00:00.0", "2013-01-02 12:00:00.0"), 
    List("2013-01-10 00:00:00.0", "2013-01-12 06:00:00.0") 
) 

//Convert the intervals list to a DataFrame 
val dfIntervals = sc.parallelize(
    filterList.map(l => (l(0),l(1))) 
).toDF("start_ts","end_ts") 

//Join both dataframes (inner join, since you only want matching rows) 
val joined = dfData.as("data").join(
    dfIntervals.as("inter"), 
    $"data.timestamp".between($"inter.start_ts", $"inter.end_ts") 
)

来源

2016-08-04 07:14:09 cheseaux

如果在列表中的任何一组点之间过滤Spark DataFrame

回答

相关问题