查看logstash 导入数据到es中，数据导入的数量

监控 logstash 将日志数据同步至 Elasticsearch 中的过程，并记录已同步的数据量；同时使用 stat 命令查询指定目录下的文件状态。

配置文件数据流的处理进度：
输入字段：
文件配置：
路径设置为：
"/home/raw_data/8_31/*.csv" 起始位置设为：'beginning' 数据库路径设为："/home/es/sincedb/apk"
类型设为：'apk'
标签设置为：['hgw', 'gather']

查看logstash处理文件进度记录：
input字段：
file配置项：
路径指定如下：
"/home/raw_data/8_31/*.csv" 起始读取位置设定为：'beginning' 数据库路径设定为："/home/es/sincedb/apk"
数据类型设定为：'apk'
相关标签包括以下几点：
'hgw'和'gather'

该过滤逻辑基于JSON格式进行配置，默认启用所有字段并执行特定数据校验操作。
其中包含以下主要功能模块：

1. 指定输入文件类型为APK时执行特定数据校验
2. 采用CSV编码方式进行字段映射
3. 定义了详细的数据字段列表
4. 设置分隔符为竖线字符
5. 设置引用字符为全角破折号
6. 如果[类型]不等于1或[ProbeOUI]不等于'YUCHUANG'时将移除上述字段

output{
if [type] == "apk"
{
elasticsearch
{
hosts => ["10.0.0.20:9200"]
index => "ana-%{type}"
document_type => "%{type}"
flush_size => 8000
idle_flush_time => 10
sniffing => true
template_overwrite => true
codec => "json"
}
}
}
1、/home/raw_data/8_31/目录下所有的.csv文件都将处理
2、第一次处理文件是从头开始
3、处理之后以json的格式输出到elasticsearch中

Logstash在文件处理过程中记录文件处理进度至sincedb中。
在示例中，将sincedb配置为/home/es/sincedb/apk；通过apk即可查看Logstash导入的处理进度。
第一列显示inode信息；第二列显示主设备编号；第三列显示次设备编号；第四列显示文件大小信息。
8589938532\ 0\ 64770\ 634171044
8589938540\ 0\ 64770\ 657879019
使用stat命令查询文件详细信息
[root@localhost 8_31]# stat /home/raw_data/8_31/*

文件："./14_26.csv"
大小：634171044 块：1238616 IO 块：4096 普通文件
设备：fd02h/64770d Inode：8589938532 硬链接：1
权限：(0644/-rw-r--r--) Uid：( 0/ root) Gid：( 0/ root)
环境：unconfined_u:object_r:home_root_t:s0
最近访问：2017-08-31 17:06:34.969528969 +0800
最近更改：2017-08-31 14:26:00.000000000 +0800
最近改动：2017-08-31 16:23:06.779714540 +0800
创建时间：-
文件："./14_31.csv"
大小：657879019 块：1284920 IO 块：4096 普通文件
设备：fd02h/64770d Inode：8589938540 硬链接：1
权限：(0644/-rw-r--r--) Uid：( 0/ root) Gid：( 0/ root)
环境：unconfined_u:object_r:home_root_t:s0
最近访问：2017-08-31 17:20:29.541469590 +0800
最近更改：2017-08-31 14:30:59.000000000 +0800
最近改动：2017-08-31 16:23:37.275712370 +0800
如果往这两个文件追加数据则将往下继续处理，而且也会将进度更新到sincedb文件中。
如果没有处理完了关闭logstash，下次再启动时则不会再从头开始处理，因为sincedb已经记录了进度，不要以为start_position => “beginning”就是每次都从头处理，如果把sincedb文件删了又会从头开始处理。