使用Kibana的动态查找

Question

我试着用卡巴纳可视化一些数据。

让我们举一个例子：

我没有几个条目可以销售一些产品，它们的事务ID作为一个事件，而它在另一个事件中的成本。

Line  1     {"event" : {"transId" : "q4h56", "prod" : "notebook"} }  
Line  2     {"event" : {"transId" : "q4h56", "cost" : 100       } }  
Line  3     {"event" : {"transId" : "rt45e", "prod" : "crayons" } }  
Line  4     {"event" : {"transId" : "rt45e", "cost" : 75        } }  
Line  5     {"event" : {"transId" : "gr56d", "prod" : "pen"     } }  
Line  6     {"event" : {"transId" : "gr56d", "cost" : 83        } }  
Line  7     {"event" : {"transId" : "uh65r", "prod" : "paper"   } }  
Line  8     {"event" : {"transId" : "uh65r", "cost" : 49        } }  
Line  9     {"event" : {"transId" : "yh5er", "prod" : "crayons" } }  
Line 10     {"event" : {"transId" : "yh5er", "cost" : 140       } }  
Line 11     {"event" : {"transId" : "23yg5", "prod" : "pen"     } }  
Line 12     {"event" : {"transId" : "23yg5", "cost" : 83        } }  

我想达到的目标是：

1:使用原木储存物：

是否可以将具有类似transId的事件合并为单个事件，如下所示：

Line  1     {"event" : {"transId" : "q4h56", "prod" : "notebook",   "cost" : 100} }  
Line  2     {"event" : {"transId" : "rt45e", "prod" : "crayons",    "cost" : 75 } }  
Line  3     {"event" : {"transId" : "gr56d", "prod" : "pen",        "cost" : 83 } }  
Line  4     {"event" : {"transId" : "uh65r", "prod" : "paper",      "cost" : 49 } }  
Line  5     {"event" : {"transId" : "yh5er", "prod" : "crayons",    "cost" : 140} }  
Line  6     {"event" : {"transId" : "23yg5", "prod" : "pen",        "cost" : 83 } }  

由于日志存储一次只解析一行，这似乎是无法实现的。

我需要存储大量全局密钥/值对，以便稍后使用，比如"%{transId}":"%{prod}"，如果再次出现相同的transId，我可以在那里插入prod，前提是它已经有了成本。

*[@metadata]对于每一行日志存储解析都是唯一的，并且在下一个解析周期中不会通过日志存储获得。如果我没有错的话！

2:使用Kibana：

是否有可能对我们销售的每一种类似产品的成本进行可视化计算。

因为成本不是用产品名称映射的。

在可视化中查看这些产品，不要提供成本。

如何映射这些，某种动态查找/连接，我觉得没有支持Kibana，再次，如果我没有错！

有什么建议吗？

同样，这只是一个示例数据，实际上使用的是具有类似情况的事件的syslog。

谢谢

Answer 1

您可以使用聚合筛选器在logstash中执行此操作。您的用例类似于文档中的例1。

如果事件的顺序是可变的，那么使用示例3。在这种情况下，将产品名称和成本存储在映射中，并event.cancel所有源事件。

Answer 2

也许Logstash的集料过滤器可以帮助您实现这一目标。

考虑到样本数据：

{"event" : {"transId" : "q4h56", "prod" : "notebook"} }
{"event" : {"transId" : "q4h56", "cost" : 100 } }
{"event" : {"transId" : "rt45e", "prod" : "crayons" } }
{"event" : {"transId" : "rt45e", "cost" : 75} }
{"event" : {"transId" : "gr56d", "prod" : "pen" } }
{"event" : {"transId" : "gr56d", "cost" : 83} }
{"event" : {"transId" : "uh65r", "prod" : "paper" } }
{"event" : {"transId" : "uh65r", "cost" : 49} }
{"event" : {"transId" : "yh5er", "prod" : "crayons" } }
{"event" : {"transId" : "yh5er", "cost" : 140 } }
{"event" : {"transId" : "23yg5", "prod" : "pen" } }
{"event" : {"transId" : "23yg5", "cost" : 83} }

下面的聚合筛选器将根据transId值聚合行。

aggregate {
    task_id => "%{[event][transId]}"
    code => "
        map['event'] ||= {}
        map['event']['transId'] ||= event.get('[event][transId]')
        map['event']['prod'] ||= event.get('[event][prod]')
        map['event']['cost'] ||= event.get('[event][cost]')
        event.cancel()
    "
    push_previous_map_as_event => true
}

这将为您提供一个具有prod和cost字段的事件。