​实时/准实时方案可以使用以下四种方式实现：

• flume+kafka+spark准实时写入ES
• logging + CMRESHandler实时写入ES
• 利用python中的Elasticsearch库实时写入ES
• pyinotify后台监控文件变化写入ES

定时写入方案则有：

• 超时重试及定时批量写入ES

接下来我将比较这几个方案的区别，及实现定时写入ES方案。

flume+kafka+spark实时写入ES

此方案流程为：通过flume采集日志，上送到kafka，接着spark程序订阅Topic并消费日志，然后写入ES。因为消息存在一定的滞后，所以叫准实时。对于数据量超大的场景，或同时需要利用spark做日志分析场景，一般采用此方案，目前公司使用的就是这样一套系统，但存在很多问题。目前随着集群上应用越来越多，资源可能不够用，导致程序偶尔会挂掉，另外kafka数据消息偶尔延时以及丢失问题，最终决定弃用本方案(项目对日志有精确要求，日志语料需要导出标注)

logging + CMRESHandler实时写入ES

另一种实时写入方案是：利用logging配合es写入模块做实时写入操作

环境安装：

pip install CMRESHandler

简易使用方式：
CMRESHandler支持的参数很多，配合logging可以方便进行日志写入管理，若想配置更简单，可以安装loguru，基本用法如下：

使用kibana查看写入结果：

注意：实际的索引默认以天为单位(可以修改)，写入内容包括主机名，ip等信息。

利用python中的Elasticsearch库实时写入ES

除了以上方案外，还可以使用python中的ElasticSearch包，调用index接口写入(当然也可以自己构造request请求写入)，首先要安装elasticsearch ：

pip install elasticsearch

写入用法如下：

使用kibana查看写入结果：

可以看出，写入内容简洁很多。

以上2种实时方案看起来不错，但也存在几个问题。

• 数据丢失问题
  若ES集群某段时间出现异常，或某个客户端节点不可用，数据存在丢失的风险
  网络异常造成访问超时，同样存在数据丢失风险
• 并发写入性能问题
  高并发写入ES库，对ES的性能要求造成很大挑战，同时会降低应用的并发能力(若使用异步线程问题倒不大)

pyinotify后台监控文件变化写入ES

为了应对高并发及网络异常情况，可以选择利用pyinotify后台监控，这样既能实时，又能在出错时将文件存在本地。

安装：

pip install pyinotify

使用方式如下：

向文件/tmp/1.log中写入内容：

echo “222” >> /tmp/1.log

程序打印：

222

可以看出，程序能实时检测到文件变化，若日志并发写入要求不是很高，则可以考虑此方案，但同时需要注意监控过程中程序中断问题，中断前需要存储pos位置，方便程序重启后读取，而不是从头开始再写入一次。

自定义超时重试请求及定时批量写入ES

综合业务(对实时性要求不高，允许短暂延迟)需求，和当前集群情况(一个ES集群，2个数据节点，2个客户端节点)，决定采用定时写入方案。实现步骤和特点如下：

• 后台进程每隔3秒检测日志目录
• 根据上次读取位置获取最新日志并写入
• 获取上次重试失败日志并再次写入

后台进程每隔3秒检测日志目录

日志目录结构为：

- logs

+ visit
- visit_20191117
- position_info

+ warn
…

+ bury
…
其中visit为访问日志，warn为告警日志，bury为埋点日志。
position_info存放上次读取的日志文件和位置信息，内容如下：
{“last_day_file_name”: “visit_20191117”, “position”: 73709}

根据上次读取位置获取最新日志并写入

这里需要注意几个问题：

• 读取日志过程中另外进程写入，会造成最后一行数据不完整的问题。因此需要做完整性校验，比如判断是不是完整的json结构。
• 写入网络异常问题。写入要重试2次，若失败，则保存下来，下次再继续写入。
• 写入过程中断程序(数据已发送)，接收写入结果失败，下次重写数据导致重复问题。因此要写入唯一性的日志id(日志写入本地时就需要确定)

自定义重试写入代码：

由于需要验证，因此在Header中加入Authorization字段，一般格式如：“Authorization: Basic jdhaHY0=”，其中Basic表示基础认证， jdhaHY0=是base64编码的"user:passwd"字符串。
同时要注意批量写入(_bulk)时，各数据行要以结尾。

唯一性的日志id
一般使用python自带的uuid来生成，使用方式也很简单：

数据完整性校验
项目中日志格式为json，因此使用eval做校验，若这行出现问题，则记录上一行的位置。

最后总结一下：
本文介绍了4种实时/准实时写入ES方案和定时写入ES方案，并比较了各种方案的使用场景和优缺点。并根据项目特点，最终选择并实现了定时写入方案(又造了个轮子。。。)，大家可以根据项目特点来实现，以上供大家参考。