hadoop - 使用 Elasticsearch 实时分析事件日志

Question

每次更改某些设备的属性时，我都会收集事件日志。为此，我决定使用:

Logstash - 我的代理 IoT 应用程序将日志以 JSON 格式发送到其中，

Elasticsearch - 用于存储数据(日志)，

Kibana - 用于数据可视化。

带有日志的 JSON 正在定期发送，其形式如下:

{"deviceEventLogs":[{"date":"16:16:39 31-08-2016","locationName":"default","property":"on","device":"Lamp 1","value":"
false","roomName":"LivingRoom"}, ... ,]}

Elasticsearch 中的单个事件条目示例如下所示:

 {
            "_index": "logstash-2016.08.25",
            "_type": "on",
            "_id": "AVbDYQPq54WlAl_UD_yg",
            "_score": 1,
            "_source": {
               "@version": "1",
               "@timestamp": "2016-08-25T20:25:28.750Z",
               "host": "127.0.0.1",
               "headers": {
                  "request_method": "PUT",
                  "request_path": "/deviceEventLogs",
                  "request_uri": "/deviceEventLogs",
                  "http_version": "HTTP/1.1",
                  "content_type": "application/json",
                  "http_user_agent": "Java/1.8.0_91",
                  "http_host": "127.0.0.1:31311",
                  "http_accept": "text/html, image/gif, image/jpeg, *; q=.2, */*; q=.2",
                  "http_connection": "keep-alive",
                  "content_length": "34861"
               },
               "date": "2016-08-08T14:48:11.000Z",
               "device": "Lamp 1",
               "property": "on",
               "locationName": "default",
               "roomName": "LivingRoom",
               "value_boolean": true
            }
 }

我的目标是创建一个带有某种仪表板的网站，在合理的时间内显示分析的数据(几分钟应该是可以接受的)，即:

显示能源消耗的历史记录并预测特征中的消耗

检测能源消耗或其他因素(如灯光或供暖使用情况)的异常

显示基于某种非复杂统计数据的建议，即“您可以将给定设备从位置 1 移动到位置 2，因为那里更需要它(比其他地方使用更密集)”等

虽然最后一点非常简单——我可以在 Elasticsearch 中使用简单的查询或聚合，然后将其与某个阈值进行比较，但前两点需要深入分析，如机器学习或数据挖掘。

目前，该系统配备了大约 50 个设备，平均每 10 秒更新一次其状态。 future 设备的数量可以增加到 50 000。假设一个事件日志有 100 字节，它可以导致 Elasticsearch 每年大约 15 TB 的数据。

一般的问题是 - 这种系统的合理解决方案/技术/架构是什么？

将我的所有日​​志存储在 Elasticsearch 中是否是一个合理的开始？

我认为 es-hadoop 库将 Elasticsearch 与 Apache Spark 一起使用，以便能够在 Spark 中使用 Mlib 处理我的数据——这是一个合理的方向吗？

我可以只使用 Elasticsearch 来存储我的所有数据并只使用 Spark 和 Mlib 来提供深入分析，还是应该考虑实现所谓的“Lambda 架构”，将 Elasticsearch 视为速度层？我对使用 Kafka、Apache Storm 的各种配置略知一二，但我不确定我是否需要它。由于该项目应该在一个月内完成，而且我是初学者，我担心复杂性，因此实现此类需要时间。

如果数据负载小 10 倍(每年大约 1.5 TB)怎么办 - 您的答案是否相同？

Answer 1

这是一个非常复杂的问题，让我试着分解一下:

您应该思考的问题

您的数据可用于查询的端到端延迟是多少？你需要实时的还是可以接受延迟？

您愿意容忍的数据丢失是多少？

您正在查看的分析/机器学习算法的准确性如何？您需要高度准确的结果还是可以接受一些不准确的结果？

您是只在完成后才需要结果还是需要某种推测性结果？

这些问题以及空间限制和数据负载增加时的延迟等常规问题应该可以帮助您确定正确的解决方案。

一般来说，这些问题可以看作是摄取 -> 处理 -> 呈现。

摄取 - 需要消息总线

通常，人们选择像 Kafka 这样的消息总线来处理来自缓慢下游消费者的背压，并提供可靠性(通过持久化到磁盘)以防止数据丢失。 Kafka 在 Spark 流、Druid firehose 支持、ES 插件等集成方面也有很好的社区支持。

处理 - 需要可扩展的计算层

这是您需要决定实时与批处理、适用的数据丢失、准确与推测结果等事项的地方。阅读 Tyler Akidau 的关于流媒体的文章 https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-101的详细解释。

人们为实时用例选择 Spark 流，一个简单的 M/R 作业应该可以解决批处理作业的问题。如果您计划流式处理作业，那么事件的窗口和 session 会使事情进一步复杂化。

演示文稿 - 需要交互式查询和快速响应

这是前端应用程序将要集成的地方，选择一个非常适合预期查询类型和所需响应准确性的工具是有意义的。

像 ES 这样的工具在搜索、过滤和分面方面表现得非常好，但在需要复杂的数学聚合时就会失败。 AFAIK ES 不像 Druid 那样支持像 HyperLogLog 这样的概率结构。

retrofit

现在您必须将您的要求与上面的每一层进行映射。

showing history of energy consumption and predicting the consumption in the feature

detecting anomalies in energy consumption or other factors like lights or heating usage

正如您所提到的，您显然需要机器学习库。 Spark 及其 MLib 支持非常棒。

showing recomendations based on some kind of not sofisticated statistics i.e. "you can move a given device from location1 to location2 because it's more needed there (more intensively used than in other place)", etc.

你甚至可以在 Spark 上使用 MLib 来做到这一点，并将建议抽取到 ES 中的单独索引甚至 Kafka 主题，你可以进一步将其归结为 HDFS 或 ES。您应该在这里小心垃圾收集，因为这可能导致数据爆炸，并且您需要在此处积极保留。此外，预先计算建议可以帮助您进行响应式操作，例如警报、推送通知，甚至来自 UI 的查询也会更快。

Assumig 100 bytes for one event log it can lead in approximation of around 15 Terabytes of data in Elasticsearch per year.

这些是任何存储系统配置的正常问题。您可以通过计算历史数据的物化 View 来优化这里，但您可以稍后再做决定，因为这可能导致过早优化。您最好先测量查询的存储和延迟，然后对容量进行追溯分析。

Is it a resonable start to store all my logs in Elasticsearch?

考虑到您的用例，非常如此。但是，如果使用 Spark 流/MLib 或批处理 MR 作业，那么您甚至可以使用愚蠢的数据存储，因为大多数计算都是事先发生的。

I consider es-hadoop library to use Elasticsearch along with Apache Spark to have an ability to process my data using Mlib in Spark - is it a resonable direction to go?

看起来您已经决定使用批处理，在这种情况下，您可以将标准 MR 或 Spark 批处理与 MLib 一起使用。如果你需要实时，你需要像 Kafka 这样的东西并使用 Spark 流。如果您对数据丢失没有意见，那么当您决定开窗/滑动间隔等时，您可以积极地保留甚至在 Spark 中。如果您对结果不准确没有意见，则可以使用概率数据结构(例如bloom filter, hyperloglog - druid 支持这个)来表示结果。

Can I use only Elasticsearch to store all my data in it and just use Spark and Mlib to provide in-depth analysis or should I consider implementing so called "Lambda Architecture" treating Elasticsearch as a Speed Layer?

我不确定您是否可以将数据从 ES 流式传输到 Spark 作业。并且 lambda 架构被过度宣传，只有在您确定您的实时层不准确并且您无法处理数据丢失/不准确的情况下才会有所帮助。否则，一个简单的 Spark 流作业从 Kafka 读取数据并泵送到 ES 就足够了。在您决定像 Lambda 这样的复杂架构之前，请考虑测量数据丢失，因为运营成本(如重复代码、需要维护的更多基础设施等)可能很高。

What if data load would be 10x smaller (around 1,5 Terabytes per year) - will your answer be the same?

我仍然更喜欢相同的架构——Kafka+Spark 流(MLib)+ES/Druid——这更容易实现，也更容易维护。