python - 优化旋转和填充

Question

他们给了我一个存储传感器读数的表，其模式为[TimeStamp, SensorKey, SensorValue]。

TimeStamp             Id           Value
2019-01-01 00:00:47   1            66.6
2019-01-01 00:00:47   2            0.66
2019-01-01 00:00:57   1            66.7
2019-01-01 00:00:57   2            0.68
2019-01-01 00:00:57   3            166.6
2019-01-01 00:01:07   3            146.6

请注意，它仅存储传感器读数的更改，精度和采样率有限，并且如果没有更改，则在上次更改后每小时重复一个值。

他们的查询意味着当传感器 Z 值通过此条件时检查传感器 A(以及 B、C、D...)的值。他们想要使用 Python 和 Spark。

因此，为了比较不同传感器的值，我获取这些传感器键的行，并将结果转换为模式 [TimeStamp、ValueOfA、...、Z 值]。

df1 = df0.groupBy("TS").pivot("Id", listOfIds).agg(F.last("Value"))

TimeStamp             Sensor1      Sensor2     Sensor3
2019-01-01 00:00:47   66.6         0.66        Null
2019-01-01 00:00:57   66.7         0.68        166.6
2019-01-01 00:01:07   Null         Null        146.6

然后我填补空白(总是向前，如果我没有旧数据来填充第一行，我会丢弃它们)。

window1hour = Window.orderBy('TS').rowsBetween(-360, 0)
# 360 = 1 hour / 0.1 Hz sampling rate.
df2 = df1
for sid in sensorIds:
    df2 = df2\
        .withColumn(sid, F.last(F.column(sid), ignorenulls=True).over(window1hour))\
        .filter(F.column(sid).isNotNull())

现在逐列比较已经很简单了。

但是与使用 pandas 进行相同的操作相比，它的速度要慢得多，以至于感觉我做错了什么。至少对于小查询来说是这样。

发生什么事了？ 当查询很大时会发生什么？

关于小型和大型:我每年拥有数千个不同的传感器和大约十亿条记录。因此，数据肯定适合一台服务器，但不适合 RAM。事实上，他们一开始只会使用一台服务器来存储数据，也许再用一台服务器来存储第二个 Spark 实例(多处理器且具有大量内存)，如果看到返回，他们希望他们会投资更多的硬件。他们将开始每天进行小查询，并且希望查询速度快。但以后他们会想做几年的查询，而且一定不能爆炸。

想法/疑问:预处理是在单线程中完成的吗？我应该自己建立并行化，还是让 Spark 处理它？我是否应该将跨年查询分解为多个跨天查询(但那我为什么要使用 Spark)？我是否可以解决 pandas 中的小查询和 Spark 中的大查询(我可以预先设置阈值)吗？

我还可以应用哪些其他改进？

Answer 1

“小”数据在 Spark 以外的工具中速度更快的情况并不罕见。 Spark 的并行功能具有相当大的开销(当然，与旧的 Map-Reduce 范例相比，这些开销非常小)。

spark 的亮点在于它能够通过添加服务器来线性扩展“大”数据。此时，开销就变得值得了，因为它会自动将工作分解到所有可用的执行器之间。

我相信让 Spark 处理并行化是理想的选择，即使只是为了简单起见。是否在另一个框架中实现“小”查询完全取决于您是否要维护两个代码路径，以及您的客户是否对它们的速度感到满意。