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

flume+kafka+spark准实时写入ESlogging + CMRESHandler实时写入ES利用python中的Elasticsearch库实时写入ESpyinotify后台监控文件变化写入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)时，各数据行要以\n结尾。

唯一性的日志id

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

数据完整性校验

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

最后总结一下：

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