网站描述怎样写,网站开发工作分解结构的树形图,ui做标注的网站,自己网站如何做关键词排名作者#xff1a;来自 Elastic Andre Luiz 将 Logstash 与 Elasticsearch 集成以实现高效的数据提取、索引和搜索的分步指南。 什么是 Logstash#xff1f;
Logstash 是一种广泛使用的 Elastic Stack 工具#xff0c;用于实时处理大量日志数据。它充当高效的数据管道#x…作者来自 Elastic Andre Luiz 将 Logstash 与 Elasticsearch 集成以实现高效的数据提取、索引和搜索的分步指南。 什么是 Logstash
Logstash 是一种广泛使用的 Elastic Stack 工具用于实时处理大量日志数据。它充当高效的数据管道将来自各种来源的信息集成到单一结构化流中。其主要功能是可靠地执行数据提取、转换和加载。
Logstash 具有多种优势尤其是其在支持多种类型的输入、过滤器和输出方面的多功能性可与各种来源和目的地集成。它实时处理数据捕获和转换信息。它与 Elastic Stack尤其是 Elasticsearch 和 Kibana的原生集成有助于数据分析和可视化。此外它还包括高级过滤器可实现高效的数据规范化、丰富和转换。 Logstash 如何工作
Logstash 由输入、过滤器和输出组成它们构成了数据处理管道。这些组件在定义数据提取流程的 .config 文件中进行配置。
输入Inputs从各种来源捕获数据。过滤器Filters处理和转换捕获的数据。输出Outputs将转换后的数据发送到定义的目的地。 每个组件最常见的类型如下所示
输入类型
文件读取各种格式文本、JSON、CSV 等的日志文件。消息队列Kafka、RabbitMQ。APIWebhook 或其他数据收集 API。数据库用于关系数据提取的 JDBC 连接。
过滤器类型
Grok用于分析和提取文本模式。Mutate修改字段重命名、转换类型、删除数据。Date将日期和时间字符串转换为可读的日期格式。GeoIP使用地理数据丰富日志。JSON解析或生成 JSON 数据。
输出类型
Elasticsearch最常见的目的地Elasticsearch 是一个搜索和分析引擎允许对 Logstash 索引的数据进行强大的搜索和可视化。Files将处理后的数据存储在本地。云服务Logstash 可以将数据发送到各种云服务例如 AWS S3、Google Cloud Storage、Azure Blob Storage进行存储或分析。数据库Logstash 可以通过特定的连接器将数据发送到各种其他数据库例如 MySQL、PostgreSQL、MongoDB 等。 Elasticsearch 的数据提取
在此示例中我们使用 Logstash 将数据提取到 Elasticsearch 中。此示例中配置的步骤将具有以下流程
Kafka 将用作数据源。Logstash 将使用数据应用 grok、geoip 和 mutate 等过滤器来构造数据。转换后的数据将发送到 Elasticsearch 中的索引。Kibana 将用于可视化索引数据。 先决条件
我们将使用 Docker Compose 创建一个具有必要服务的环境Elasticsearch、Kibana、Logstash 和 Kafka。Logstash 配置文件名为 logstash.conf将直接挂载到 Logstash 容器中。下面我们将详细介绍配置文件的配置。
这是 docker-compose.yml
version: 3.8
services:elasticsearch:image: docker.elastic.co/elasticsearch/elasticsearch:8.16.2container_name: elasticsearch-8.16.2environment:- node.nameelasticsearch- xpack.security.enabledfalse- discovery.typesingle-node- ES_JAVA_OPTS-Xms1024m -Xmx1024mports:- 9200:9200networks:- shared_networkkibana:image: docker.elastic.co/kibana/kibana:8.16.2container_name: kibana-8.16.2restart: alwaysenvironment:- ELASTICSEARCH_URLhttp://elasticsearch:9200ports:- 5601:5601depends_on:- elasticsearchnetworks:- shared_networklogstash:image: docker.elastic.co/logstash/logstash:8.16.2container_name: logstash-8.16.2volumes:- ./logstash.conf:/usr/share/logstash/pipeline/logstash.confports:- 5044:5044depends_on:- elasticsearchnetworks:- shared_networkzookeeper:image: confluentinc/cp-zookeeper:latestcontainer_name: zookeeperenvironment:ZOOKEEPER_CLIENT_PORT: 2181ZOOKEEPER_TICK_TIME: 2000ports:- 2181:2181networks:- shared_networkkafka:image: confluentinc/cp-kafka:latestcontainer_name: kafkadepends_on:- zookeeperenvironment:KAFKA_BROKER_ID: 1KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1ports:- 9092:9092networks:- shared_networknetworks:shared_network:
如上所述将定义 Logstash 管道在此步骤中我们将描述输入、过滤器和输出配置。
将在当前目录docker-compose.yml 所在的位置中创建 logstash.conf 文件。在 docker-compose.yml 中本地文件系统上的 logstash.conf 文件将安装在容器内的路径 /usr/share/logstash/pipeline/logstash.conf 中。 Logstash 管道配置
Logstash 管道分为三个部分输入、过滤器和输出。
输入定义数据的使用位置在本例中为 Kafka。过滤器对原始数据进行转换和结构化。输出指定处理后的数据发送到的位置在本例中为 Elasticsearch。
接下来我们将详细配置每个步骤。 输入配置
数据源是 Kafka 主题要使用该主题的数据需要配置 Kafka 输入插件。以下是 Logstash 中 Kafka 插件的配置我们定义
bootstrap_serversKafka 服务器的地址。topics要使用的主题的名称。group_id消费者组标识符。
input {kafka {bootstrap_servers kafka:9092topics [logs]group_id logstash-consumer}
}
这样我们就可以接收数据了。 过滤器配置
过滤器负责转换和构造数据。让我们配置以下过滤器 Grok 过滤器
从非结构化数据中提取结构化信息。在本例中它提取时间戳、日志级别、客户端 IP、URI、状态和 JSON 负载。
grok {match {message %{TIMESTAMP_ISO8601:timestamp},%{WORD:log_level},%{IP:client_ip},%{URIPATH:uri},%{NUMBER:status}}
}
示例日志
2025-01-05 16:30:15,INFO,69.162.81.155,/api/products,200,{user_id:123,region:US}
提取字段
timestamp提取日期和时间例如2025-01-05T16:30:15。log_level捕获日志级别例如INFO、ERROR。client_ip捕获客户端 IP 地址例如69.162.81.155。uri捕获 URI 路径例如/api/products。status捕获 HTTP 状态码例如200。 日期过滤器
将时间戳字段转换为 Elasticsearch 可读的格式并将其存储在 timestamp 中。
date {match [timestamp, yyyy-MM-dd HH:mm:ss]target timestamp} GeoIP 过滤器
接下来我们将使用 geoip 过滤器根据 client_ip 字段的值检索地理信息例如国家、地区、城市和坐标。 geoip {source client_iptarget geoip } Mutate 过滤器
变异过滤器允许对字段进行转换。在本例中我们将使用它的两个属性
remove_field删除时间戳和消息字段因为它们不再需要。convert将状态字段从字符串转换为整数。 输出配置
输出定义转换后的数据将发送到何处。在本例中我们将使用 Elasticsearch。
output {elasticsearch {hosts [http://172.21.0.1:9200]index webapp_logs}
}
现在我们已经定义了配置文件。以下是完整文件
input {kafka {bootstrap_servers kafka:9092topics [logs]group_id logstash-consumer}
}filter {grok {match {message %{TIMESTAMP_ISO8601:timestamp},%{WORD:log_level},%{IP:client_ip},%{URIPATH:uri},%{NUMBER:status}}}date {match [timestamp, yyyy-MM-dd HH:mm:ss]target timestamp}geoip {source client_iptarget geoip}mutate {remove_field [timestamp, message]convert { status integer }}
}output {elasticsearch {hosts [http://172.21.0.1:9200]index webapp_logs}
} 发送和提取数据
容器运行时我们可以开始向主题发送消息并等待数据被索引。首先如果尚未创建主题topic请创建主题。
docker exec -it kafka kafka-topics --create --topic logs --bootstrap-server kafka:9092 --partitions 1 --replication-factor 1
要发送消息请在终端中执行以下命令
docker exec -it kafka kafka-topics --create --topic logs --bootstrap-server kafka:9092 --partitions 1 --replication-factor 1
要发送的消息
2025-01-05 16:30:15,INFO,69.162.81.155,/api/products,200,{user_id:123,region:US}
2025-01-05 16:31:02,ERROR,104.101.21.255,/api/orders,500,{user_id:124,region:BR}
2025-01-05 16:32:45,INFO,103.244.145.255,/api/cart,404,{user_id:125,region:DE}
要查看索引数据请转到 Kibana 索引成功完成后我们可以在 Kibana 中查看和分析数据。映射和索引过程可确保字段根据 Logstash 中定义的配置进行结构化。 结论
通过提供的配置我们使用 Logstash 创建了一个管道用于在具有 Elasticsearch 和 Kafka 的容器化环境中索引日志。我们探索了 Logstash 使用 grok、date、geoip 和 mutate 等过滤器处理消息的灵活性从而构建了数据以供在 Kibana 中进行分析。此外我们还演示了如何配置与 Kafka 的集成以使用消息并使用它们来处理和索引数据。 参考
Logstash https://www.elastic.co/guide/en/logstash/current/index.html Logstash Docker https://www.elastic.co/guide/en/logstash/current/docker.html GeoIp Plugin https://www.elastic.co/guide/en/logstash/current/plugins-filters-geoip.html Mutate Plugin https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html Grok Plugin https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html Kafka Plugin https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html
想要获得 Elastic 认证了解下一期 Elasticsearch 工程师培训何时举行
Elasticsearch 包含许多新功能可帮助你针对自己的用例构建最佳搜索解决方案。深入了解我们的示例笔记本以了解更多信息开始免费云试用或立即在本地机器上试用 Elastic。 原文How to ingest data to Elasticsearch through Logstash - Elasticsearch Labs
