使用Logstash处理数据 配置处理任务 1. 在云搜索服务实例创建页面选择Logstash实例管理,选择已经加装的Logstash实例。 2. 选择“管道管理”,选择“新建管道”。 3. 在“管道配置”窗口中,配置相关的管道配置。 python input { kafka { autooffsetreset > "latest" bootstrapservers > "ip:port,ip:port,ip:port" consumerthreads > 88 groupid > "yourgroupid" topics > ["yourkafkatopic"] codec > json } } filter { } output { elasticsearch { hosts > ["ip:port"] index > "yourindexname%{+YYYY.MM.dd}" user > "yourusername" password > "yourpassword" } } input部分用于定义数据的来源,即Logstash从哪里获取数据。这里以Kafka为例: autooffsetreset:会决定从哪个位置开始消费消息 bootstrapservers:Kafka集群的地址 consumerthreads:控制 Kafka 消费者线程的数量 groupid:消费者组ID topics:Kafka的主题列表 codec:指定数据格式 filter部分可以进行消息处理、数据解析等操作。根据实际业务规则来处理。 output部分用于将处理后的数据推送到外部系统(如数据库、消息队列、搜索引擎等)。这里以Elasticsearch为例: hosts:指定Elasticsearch的地址。 index:指定存储到Elasticsearch的索引名称,可以使用日期变量%{+YYYY.MM.dd} 来动态生成基于日期的索引。 user/password:提供Elasticsearch用户名和密码 配置完点击“预校验”,预校验会使用Logstash的语法功能来检测基础的语法,不检测逻辑和连通性。 预校验通过后可以执行后续操作。 4. 在“参数配置”中,配置相关的参数。 参数 说明 管道名称 只能包含字母、数字、中划线或下划线,且必须以字母开头,限制430个字符,实例内唯一 管道描述 限制长度50 管道工作线程数 控制Logstash同时处理事件的线程数 管道批处理大小 单个线程从input部分收集的最大事件数 管道批处理延迟 不满足批处理大小最大等待时间 队列类型 memory模式、persisted模式 memory:内存队列 persisted:磁盘队列 队列最大字节数 persisted模式生效 队列最大存储大小 队列检查点写入数 persisted模式生效 队列检查点数量 5. 启动处理任务 配置完参数后点击“保存并部署”。就启动了读取Kafka中的数据转换后写入到Elasticsearch的管道任务。 等到管道“运行中”可以尝试配置的Kafka的topics中导入一些数据,例如使用Filebeat或其他程序。 可以通过Curl命令查看logstash中配置的索引名,在Elasticsearch实例是否有数据进来。 plaintext curl " 6. 停止任务 当任务执行结束后,可以在“管道管理”页面,点击“停止”,会执行停止管道操作。