项目背景
自从2016年AlphaGo击败人类围棋选手，将时代带入了人工智能时代。无论机器机器学习还是深度学习，都依赖于数据。所以我们不得不回顾下这个10年在IT业中重要的名词。
云计算（Cloud）：对云计算的定义有多种说法。现阶段广为接受的是美国国家标准与技术研究院定义：云计算是一种按使用量付费的模式，这种模式提供可用的、便捷的、按需的网络访问， 进入可配置的计算资源共享池（资源包括网络，服务器，存储，应用软件，服务），这些资源能够被快速提供，只需投入很少的管理工作，或与服务供应商进行很少的交互。
大数据（BigData）：对于大数据的定义业界通常用4个V来解释 (即Volume、Variety、Value、Velocity)，数据体量大，数据类型多，价值密度低，数据产生快。随着设备日志数据的剧增，日志数据本身也渐渐的往大数据的形态靠拢。
物联网（IoT）：从最初在1999年提出：即通过射频识别（RFID）到现在物联网（Internet of Things）,把所有物品通过信息传感设备与互联网连接起来，进行信息交换，即物物相息，以实现智能化识别和管理。
人工智能（AI）：意图了解智能的实质，并生产出一种新的能以人类智能相似的方式做出反应的智能机器，该领域的研究包括机器人、语言识别、图像识别、自然语言处理和专家系统等。
项目内容
会使用Elastic公司的三款开源工具来进行IoTHub的构建。Elastic提供Logstash，Elasticsearch，Kibana这3款开源工具。其中Logstash为数据收集，转换，传输的工具，Elasticsearch为数据存储，搜索，分析的工具，Kibana为数据可视化展示的工具。
系统架构
Elasticsearch是分布式数据库，所以建立多台服务器做ES集群。Logstash主要起到传输作用，构建并行服务器。Kibana主要用于显示，构建单台。智能分析通过Python来进行。传感器部分通过Raspberry Pi 3等
硬件准备清单
为了简便，实验环境在Windows10中进行，真实环境可以考虑Windows Server 2016等服务器版本。
物理机：
CPU:I7 4770K,内存：32GB，硬盘：2块SSD 256GB，显卡：nvidia 1070 8G
虚拟机：
Server1：Windows 2012R2.（ElasticSearch,LogStash,Kibana）,IP:192.168.1.201
Server2: Windows 2012R2.（ElasticSearch,LogStash）,IP:192.168.1.202
Server3: Windows 2012R2.（ElasticSearch）,IP:192.168.1.203
软件准备清单
Windows环境下安装ELK需要准备软件 Java8安装包（不要安装Java10！,不要安装Java10！，不要安装Java10！目前Logstash还不支持Java10.重要的要说3边） ElasticSearch安装包，LogStash安装包，Kibana安装包，nssm安装包。 另外还有node安装包与head插件（之后会说明）
第一步JAVA安装JAVA安装
Java8下载地址，请下载Windows x64版，目前为（jdk-8u171-windows-x64.exe）
http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
按默认安装。分别在虚拟Server1到3中安装。
第二步ElasticSearch安装
1.安装ELK首先需要到
https://www.elastic.co/cn/products
下载软件。目前为6.2.4版本建议使用MSI版本软件包
2.首先在Server1中安装第一台ElasticSearch，双击MSI软件包安装。
之后向导中不用选插件。下一步进行安装。
3.修改ES配置文件
修改以下配置文件
C:\IotProject\elasticsearch\Node\config\elasticsearch.yml
bootstrap.memory_lock: false
cluster.name: IotProject
network.host: 192.168.1.201 //绑定本地IP，默认尾127.0.0.1
http.port: 9200
http.cors.enabled: true //head插件用
http.cors.allow-origin: "*" //head插件用
node.data: true
node.ingest: true
node.master: true
node.max_local_storage_nodes: 1
node.name: ELKIOTSERVER01
path.data: C:\IotProject\elasticsearch\Node\data
path.logs: C:\IotProject\elasticsearch\Node\logs
transport.tcp.port: 9300
discovery.zen.ping.unicast.hosts: ["192.168.1.201","192.168.1.202","192.168.1.203"] //集群节点所有IP地址
discovery.zen.minimum_master_nodes: 2 //宣告master节点数，一般是节点数/2+1
细节可以参考以下URL https://blog.csdn.net/zxf_668899/article/details/54582849
4.Head插件安装（监控ElasticSearch状态）
参考以下URL安装
https://blog.csdn.net/u012270682/article/details/72934270
elasticsearch 5以上版本安装head需要安装node和grunt
下载地址：https://nodejs.org/en/download/ 根据自己系统下载相应的msi，双击安装。
确认node版本，并安装grunt，用npm install -g grunt -cli命令
确认grunt版本，用grunt -version命令
完成后，修改elasticsearch.yml配置文件
http.cors.enabled: true //head插件用
http.cors.allow-origin: “*” //head插件用
重启服务器。并到https://github.com/mobz/elasticsearch-head 下载zip文件，解压缩到文件夹
进入解压缩文件夹，修改Gruntfile.js文件
cmd进入E:\elasticsearch-5.4.1\elasticsearch-head-master文件夹
执行 npm install
安装完成查看结果192.168.1.201:9100确认结果。目前没数据的话，如图
安装完成执行grunt server 或者npm run start
或head解压缩目录下创建该命令bat文件
6.按上述2-3内容，在server2，与server3中安装。其他ES集群节点 其他节点node.data: true，node.ingest: true，node.master: true。 但是head插件用配置不用追加。
ElasticSearch介绍
使用ElasticSearch搜索引擎，如图所示，从集群（Cluster），节点（Node），索引（Index），分片（Shard）的4个内容
●集群（Cluster）:
集群就是一个或多个节点存储数据，其中一个节点为主节点，这个主节点是可以通过选举产生的，并提供跨节点的联合索引和搜索的功能。集群有一个唯一性标示的名字，默认是elasticsearch，集群名字很重要，每个节点是基于集群名字加入到其集群中的。因此，确保在不同环境中使用不同的集群名字。一个集群可以只有一个节点。强烈建议在配置elasticsearch时，配置成集群模式。
目的:建立一组拥有相同集群名的节点集合，它们协同作业并共享数据并提供
故障转移和弹性扩展，一个集群也可以由一个单一节点创建。
●节点（Node）:
节点就是一台单一的服务器，是集群的一部分，存储数据并参与集群的索引和搜索功能。像集群一样，节点也是通过名字来标识，默认是在节点启动时随机分配的字符名。当然啦，你可以自己定义。该名字也蛮重要的，在集群中用于识别服务器对应的节点
目的 :运行Elasticsearch的实例。节点间通过网络进行通讯协作作业（默认是TCP 9300端口）。
集群中会有一台包涵Master模式的节点，当Master模式节点故障后，会由可升级Master节点中选择出来，并提升为Master。
·默认节点（包括可升级Master，存储Data，聚合Query）
·Master节点:集群中用于控制其他节点的工作。
·数据节点:仅存放分片数据的节点
·聚合查询节点:对外部应用程序的查询进行响应。无数据和Master
●索引（Index）:
索引是有几分相似属性的一系列文档的集合。如nginx日志索引、syslog索引等等。索引是由名字标识，名字必须全部小写。这个名字用来进行索引、搜索、更新和删除文档的操作。
索引相对于关系型数据库的库。
目的:存储关联数据的地方。实际上，索引只是一个用来指向一个或多个分片(shards)的逻辑命名空间。
●类型（Type） 在一个索引中，可以定义一个或多个类型。类型是一个逻辑类别还是分区完全取决于你。通常情况下，一个类型被定于成具有一组共同字段的文档。如ttlsa运维生成时间所有的数据存入在一个单一的名为logstash-ttlsa的索引中，同时，定义了用户数据类型，帖子数据类型和评论类型。 类型相对于关系型数据库的表。
●文档（Document） 文档是信息的基本单元，可以被索引的。文档是以JSON格式表现的。 在类型中，可以根据需求存储多个文档。 虽然一个文档在物理上位于一个索引，实际上一个文档必须在一个索引内被索引和分配一个类型。 文档相对于关系型数据库的列。
●分片（Shard）:
在实际情况下，索引存储的数据可能超过单个节点的硬件限制。如一个十亿文档需1TB空间可能不适合存储在单个节点的磁盘上，或者从单个节点搜索请求太慢了。为了解决这个问题，elasticsearch提供将索引分成多个分片的功能。当在创建索引时，可以定义想要分片的数量。每一个分片就是一个全功能的独立的索引，可以位于集群中任何节点上。
分片的两个最主要原因：
a、水平分割扩展，增大存储量
b、分布式并行跨分片操作，提高性能和吞吐量
分布式分片的机制和搜索请求的文档如何汇总完全是有elasticsearch控制的，这些对用户而言是透明的。
网络问题等等其它问题可以在任何时候不期而至，为了健壮性，强烈建议要有一个故障切换机制，无论何种故障以防止分片或者节点不可用。
为此，elasticsearch让我们将索引分片复制一份或多份，称之为分片副本或副本。
副本也有两个最主要原因：
高可用性，以应对分片或者节点故障。出于这个原因，分片副本要在不同的节点上。
提供性能，增大吞吐量，搜索可以并行在所有副本上执行。
总之，每一个索引可以被分成多个分片。索引也可以有0个或多个副本。复制后，每个索引都有主分片(母分片)和复制分片(复制于母分片)。分片和副本数量可以在每个索引被创建时定义。索引创建后，可以在任何时候动态的更改副本数量，但是，不能改变分片数。
默认情况下，elasticsearch为每个索引分片5个主分片和1个副本，这就意味着集群至少需要2个节点。索引将会有5个主分片和5个副本(1个完整副本)，每个索引总共有10个分片。
每个elasticsearch分片是一个Lucene索引。一个单个Lucene索引有最大的文档数LUCENE-5843, 文档数限制为2147483519(MAX_VALUE – 128)。 可通过_cat/shards来监控分片大小。
目的 :提供 最小级别工作单位,它只是保存了索引中所有数据的一部分。分片作为数据容器，日志数据存储在分片之中，然后分片被动态的分配到集群的各个节点，随着集群的扩容或缩小，Elasticsearch会将分片自动在节点中进行迁移，保证集群的动态平衡。
针对每个索引，分为0-4的5个分片，且有1组主分片与1组复制片。也就是供有10个分片，分在在3台主机中。
其原因为分片按特性会分为:
·主分片(primary shard) : 索引创建时会将主分片的数量固定了，但是复制分片的数量可以之后进行调整。
·复制分片(replica shard): 复制分片是主分片的副本，当主分片所在主机故障时，可以继续提供读请求，如搜索或者从别的分片中读取文档。
通过触发分片访问的概率，来提升ElasticSearch集群的性能。
（1）集群状态： green：主分片、副本分片都存在 yellow：分片丢失，但每个分片至少存在一个主分片或副本分片 red：分片丢失，主分片和副本分片都丢失
（2）Lucene的核心组件： 索引(Index)：类比数据库(database) 类型(Type)：类比表(table) 文档(Document)：类比行(row)
（3）Elasticsearch提供了RESTful的API接口，可以通过http协议与其进行交互
语法：curl -X
<BODY>：json格式的请求主体
<VERB>：GET（获取），POST（修改），PUT（创建），DELETE（删除）
<PATH>：/index_name/type/Document_ID/
特殊PATH：/_cat, /_search, /_cluster
创建文档：-XPUT -d '{"key1": "value1", "key2": value, ...}'
/_search：搜索所有的索引和类型；
/INDEX_NAME/_search：搜索指定的单个索引；
/INDEX1,INDEX2/_search：搜索指定的多个索引；
/s*/_search：搜索所有以s开头的索引；
/INDEX_NAME/TYPE_NAME/_search：搜索指定的单个索引的指定类型；
curl -XGET 'http://192.168.1.201:9200/_cluster/health?pretty=true'
curl -XGET 'http://192.168.1.201:9200/_cluster/stats?pretty=true'
curl -XGET 'http://192.168.1.201:9200/_cat/nodes?pretty'
curl -XGET 'http://192.168.1.201:9200/_cat/health?pretty'
第三部LogStash安装
1.将下载的LogStash文件解压缩到制定的目录中。同时在bin目录中，创建一个启动bat
C:\IotProject\Logstash\6.2.4\bin该目录下创建。
Bat文件如下
logstash -f conf\IotProject.conf
其中 -f为执行conf文件。-e为直接执行命令行中的配置程序。一般用-f
2.创建conf配置文件。
（1）IotProject.conf
input {
tcp{
port => 5000
codec => json
}
}
#filter {
#}
output {
elasticsearch{
hosts => ["192.168.1.201:9200","192.168.1.202:9200","192.168.1.203:9200"]
index => "persioncheck-%{+YYYY-MM-dd}"
document_type => "persioncheck"
}
}
（2）debug用conf
input {
tcp{
port => 5000
codec => json
}
}
#filter {
#}
output {
stdout{
codec => rubydebug
}
}
2.使用nssm将logstash作为windows服务启动
到http://www.nssm.cc/download 下载最新版本nssm并拷贝到项目目录。
参考 https://www.cnblogs.com/TianFang/p/7912648.html 设定logstash服务，nssm install logstash命令
path为logstash刚才创建的bat文件。
Dependencies为依赖elasticsearch，并填写es的服务名称。
LogStash介绍
1.Logstash日志管道工具进行构建。 使用其中的Input组件、Filter组件、Output组件作为构造块进行设计，
input {
tcp{
port => 5000 //TCP端口
codec => json //将数据转为json
}
}
output {
elasticsearch{
hosts => ["192.168.1.201:9200","192.168.1.202:9200","192.168.1.203:9200"] //写入到3台es服务器。IP都要
index => "persioncheck-%{+YYYY-MM-dd}" //索引的名称 - 日期
document_type => "persioncheck" //文档类型名称
}
}
内容可以参考URL https://www.cnblogs.com/moonlightL/p/7760512.html
2.filter功能
如图主要有以下功能
部分举例（本代码中无使用filter）
filter {
if [type] == "WindowsEventLog" {
grok {
match => [ "host", "^(?<host2>[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]):.*" ]
}
mutate {
replace => [ "host", "%{Hostname}" ]
}
mutate {
remove_field => [ "host2" ]
}
mutate {
lowercase => [ "EventType", "FileName", "Hostname", "Severity" ]
}
mutate {
rename => [ "Hostname", "source_hostname" ]
}
mutate {
gsub => ["source_hostname","\.example\.com",""]
}
date {
match => [ "EventTime", "YYYY-MM-dd HH:mm:ss" ]
}
mutate {
rename => [ "Severity", "evt_severity" ]
rename => [ "SeverityValue", "evt_severityvalue" ]
rename => [ "Channel", "evt_channel" ]
rename => [ "SourceName", "evt_program" ]
rename => [ "SourceModuleName", "nxlog_evt" ]
rename => [ "Category", "evt_category" ]
rename => [ "EventID", "evt_ID" ]
rename => [ "RecordNumber", "evt_recordnumber" ]
rename => [ "ProcessID", "evt_processID" ]
}
if [SubjectUserName] =~ "." {
mutate {
replace => [ "AccountName", "%{SubjectUserName}" ]
}
}
if [TargetUserName] =~ "." {
mutate {
replace => [ "AccountName", "%{TargetUserName}" ]
}
}
if [FileName] =~ "." {
mutate {
replace => [ "evt_category", "%{FileName}" ]
}
}
mutate {
lowercase => [ "AccountName", "evt_channel" ]
}
mutate {
remove => [ "SourceModuleType", "EventTimeWritten", "EventReceivedTime", "EventType" ]
}
if [evt_program] == "WSH"{
grok{
match => ["message",".*scriptname\s(?<script_name>\b\w*\b)\s.*scriptID\s(?<script_ID>\b\w*\b)\s.*scriptresult\s(?<script_result>\b\w*\b)\s.*scriptmessage\s(?<script_message>.*)"]
}
mutate {
replace => [ "message", "%{script_message}" ]
}
mutate {
remove_field => [ "script_message" ]
}
}
}
if [type] == "DDWRT"{
grok {
match => ["message" ,".*>(?<msMonth>.*)\s(?<msDay>.*)\s(?<msTimes>.*)\skernel:\s(?<Action>\b\w*\b)\s.*SRC=(?<from_srcip>[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9])\sDST=(?<to_dstip>[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]).*PROTO=(?<protocol>\b\w*\b)\sSPT=(?<from_spt>\b\w*\b)\sDPT=(?<to_dpt>\b\w*\b).*"]
}
grok {
match => ["message" ,".*>(?<EventTime>.*)\skernel:\s(?<Action>\b\w*\b)\s.*SRC=(?<from_srcip>[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9])\sDST=(?<to_dstip>[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]\.[0-2]?[0-9]?[0-9]).*PROTO=(?<protocol>\b\w*\b)\sSPT=(?<from_spt>\b\w*\b)\sDPT=(?<to_dpt>\b\w*\b).*"]
}
grok {
match => ["@timestamp" ,"(?<stampYear>.*)-.*"]
}
mutate {
add_tag => [ "EventTime" ]
replace => [ "EventTime", "%{msMonth} %{msDay} %{msTimes}" ]
}
date {
match => [ "EventTime", "MMM dd HH:mm:ss","MMM d HH:mm:ss" ]
}
}
}
(1)Grok过滤模块
目的:解析任意文本，将非结构日志数据结构化的模块。
分析:通过子函数Match调用实现正则表达式匹配来结构化日志数据。
(2)Date处理模块
目的:将日志数据中的时间返回给“@timestamp”。
分析 :它是一种标准的 ISO8601 的时间戳，其结果类似于
“2015 -08 -16T00:49:32.000Z” 。 通过子函数Match调用完成时间格式的
匹配来反馈数据。时间格式为“YYYY-MM-dd HH:mm:ss”
(3) Mutate 处理模块
目的:对主机日志 数据进行重命名、 删除替换修改。
分析:通过子函数调用
Replace :数据 替换，通过 替换，通过 %{ 结构字段名 }将 host 替换 成 hostname 字段数据 。
Lowercase :将”EventType”, Hostname”, Severity” “EventType”, Hostname”, Severity” “EventType”, Hostname”, Severity” 数据 改为小写字符串。
Rename :重命名 结构字段的名称。 将”Hostname” 重命名为 “source_hostname” 。
Gsub:替换日志数据中的字符串格式。将”source_hostname”中的域名
如”.ComingSpace.local”替换为空。
Remove:删除日志数据中的结构字段。删除”SourceModuleType”、
“EventTimeWritten”、 “EventReceivedTime”字段及数据。
第四步kibana的安装
将kibana拷贝到文件夹，并制作bat文件。 用nssm将kibana作为服务启动。
kibana需要修改配置kibana.yml
server.port: 5601
server.host: “192.168.1.201”
elasticsearch.url: “http://192.168.1.201:9200”
然后打开URL
192.168.1.201：5601
kibana今后可以绑定地图，根据gps做应用。本次未进行实验。
第五步应用传感器
我们通过电脑结合opencv来对人脸进行检查，并提交数据给ELK
我们需要安装python-logstash。可以直接通过python传输数据
https://pypi.org/project/python-logstash/
通过命令行，pip install python-logstash安装
然后编写代码
import logging
import logstash
import cv2
import time
host = '192.168.1.201'
test_logger = logging.getLogger('python-logstash-logger')
test_logger.setLevel(logging.INFO)
test_logger.addHandler(logstash.TCPLogstashHandler(host, 5000, version=1)
cap=cv2.VideoCapture(0)
face_cascade = cv2.CascadeClassifier("C:/OneDrive/PythonCode/Configfile/OpenCV/haarcascade_frontalface_default.xml")
while (True):
# Capture frame-by-frame
ret, frame = cap.read()
# Our operations on the frame come here
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
# The detected objects are returned as a list of rectangles
faces = face_cascade.detectMultiScale(gray, 1.3, 5)
for (x, y, w, h) in faces:
cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2)
# Display the resulting frame
extra = {
'HavePersion': True,
}
test_logger.info('PersionTest', extra=extra)
cv2.imshow('frame', frame)
time.sleep(1)
if cv2.waitKey(1) & 0xFF == ord('q'):
break
# Closes video file or capturing device.
cap.release()
cv2.destroyAllWindows()
确认kibana是否已经有数据进行传入。本次实验到这里。谢谢大家。