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

Question

我有一个 spark 数据框（使用 scala 接口），它包含时间戳、资产（字符串）、标签（字符串）和值（双精度）列。这是它的摘录：

+--------------------+-----+--------+-------------------+
|           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 的大小为 2，并保存感兴趣间隔的开始和结束时间。例如：

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

持有两个兴趣区间：一个是从 2013-01-02 的午夜到 12:00，另一个是从 2013-01-10 的午夜到 2013-01-12 的 6:00

这是我的问题：如何将数据帧过滤为 return 值，以便时间戳位于任何间隔内？对于任何一个间隔，我都可以做到

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

因为我不知道 Array 中有多少元素（我有多少间隔），所以我不能只使用一长串过滤器。

对于上面的示例，我想保留第 4、6 和 9 行。

我现在拥有的是 Array 的循环，并且正在为每个循环获取适当的子集。但是，这可能比将所有内容都放在一个大过滤器中要慢，对吗？

Answer 1

您可以将时间戳列表转换为 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")
)

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

Filter Spark DataFrame if between any set of points in list

performance

scala

apache-spark

spark-dataframe