关于使用 parquet 处理时间序列数据的问题答案

【问题标题】：Question about using parquet for time-series data关于使用 parquet 处理时间序列数据的问题
【发布时间】：2019-12-05 19:04:19
【问题描述】：

我正在探索以可扩展且经济高效的方式存储来自传感器的大量数据（时间序列数据）的方法。

目前，我正在为每个传感器编写一个按日期分区的 CSV 文件，因此我的文件系统层次结构如下所示：

client_id/sensor_id/year/month/day.csv

我的目标是能够对此数据执行 SQL 查询（通常为特定客户端/传感器获取时间范围，执行聚合等）我尝试将其加载到 Postgres 和 timescaledb，但是体积太大，查询速度过慢。

我现在正在尝试使用Spark 和Parquet 文件来执行这些查询，但是我有一些问题我无法从我对这个主题的研究中得到答案，即：

我正在将这些数据转换为 parquet 文件，所以我现在有这样的东西：

client_id/sensor_id/year/month/day.parquet

但我担心的是，当Spark 加载包含许多Parquet 文件的顶层文件夹时，行组信息的元数据并没有像我使用包含所有数据的单个拼花文件那样优化，由@ 分区987654329@。这是真的？还是拥有多个 parquet 文件或单个分区 Parquet 文件是否相同？我知道镶木地板文件在内部存储在我正在使用的文件夹层次结构中，但我不清楚这如何影响文件的元数据。

我无法做到这一点的原因是我不断接收新数据，据我了解，由于页脚元数据的工作性质，我无法附加到镶木地板文件。它是否正确？现在，我只需将前一天的数据转换为 parquet 并为每个客户端的每个传感器创建一个新文件。

谢谢。

【问题讨论】：

你每次写的大小是多少？
这很难说，因为数据摄入是可变的，但给你一些更多的背景信息：传感器数据消息包含多个读数 [{sensor_id:sensor_value}, ..., {sensor_id:sensor_value}]，但是 id 的数量和种类是不可预测的。我将它们推出，以便表具有固定的架构（时间戳、id、值）。所以每条消息最多可以有 30 行，我每秒收到数千条消息。我用 timescaledb 从 kafka 摄取数据做了一个小实验，为期一周，产生了 28 亿行。
有趣的东西。不完全是我所参与的（还）。

标签： apache-spark time-series parquet

【解决方案1】：

您可以将 Structured Streaming 与 kafka 一起使用（因为您已经在使用它）来实时处理您的数据并以 parquet 格式存储数据。而且，是的，您可以将数据附加到镶木地板文件中。使用 SaveMode.Append ，例如 df.write.mode('append').parquet(path)

您甚至可以按小时对数据进行分区。客户端/传感器/年/月/日/小时，这将进一步为您提供查询时的性能改进。您可以根据要对数据运行的查询类型基于系统时间或时间戳列创建小时分区。

如果您选择基于时间戳列进行分区，您可以使用watermaking 处理延迟记录。

希望这会有所帮助！

【讨论】：

你好 wypul。我以前使用过带有 spark 的结构化流。我不认为数据转换步骤是我现在的瓶颈。我知道您可以附加到镶木地板，我的问题是是否会影响性能，因为行组需要数据来预先计算元数据。查询不是按小时粒度完成的，所以我想进一步划分分区不会带来任何好处。无论如何，我都会尝试您的建议，以将转换步骤与我现在所拥有的进行比较。谢谢
您好@mlashcorp 您有时间分享您的最终解决方案吗？ :)

【解决方案2】：

我可以分享我在 AppsFlyer 使用的经验和技术堆栈。我们有大量数据，每天大约有 700 亿个事件。

我们用于近实时分析的时间序列数据存储在 Druid 和 Clickhouse 中。 Clickhouse 用于保存最近两天的实时数据；德鲁伊（0.9）无法管理它。 Druid 保存着我们每天通过 Hadoop 填充的其余数据。

Druid 是一个合适的人选，以防您每天或每小时不需要行数据而是预先聚合的数据。我建议你给 Clickhouse 一个机会，它缺少文档和示例，但运行起来既健壮又快速。

另外，您可以看看 Apache Hudi。

【讨论】：