Logstash
Logstash is a powerful data processing pipeline tool which is a part of ELK stack (ElasticSearch, Logstash, kibana).
It is mainly used for Data Ingestion and Transformation.
Logstash collects and transforms logs, ElasticSearch stores and indexes them for fast search and Kibana provides dashboards and visualization. It is widely used for centralized logging and troubleshooting in distributed systems and microservices architectures.
Logtash helps collect, parse and transform data before sending it to a destination like ElasticSearch, Kafka or even a file.
Core Components
- Input — Defines where the data comes from (e.g. log files, Kafka,syslog).
input
{
file {
path => “/var/log/syslog”
start_position => “beginning”
}
} - Filter —This is a Logstash filter configuration used to parse unstructured logs into structured fields before sending them to Elasticsearch.
filter {
grok {
match => {
“message” => “%{SYSLOGTIMESTAMP:timestamp} %{WORD:loglevel} %{GREEDYDATA:message}”
}
}
date {
match => [“timestamp”, “MMM d HH:mm:ss”, “MMM dd HH:mm:ss”]
}
}
3. Output — Defines where to send the processed data (e.g., Elasticsearch, stdout).
- output {
elasticsearch {
hosts => [“http://localhost:9200"]
index => “syslog-%{+YYYY.MM.dd}”
}
Use Cases
- Centralized logging from multiple sources
- Log enrichment and transformation
- Routing logs to multiple systems (e.g., Kafka + Elasticsearch)
- Real-time data transformation
Example: Kafka to Elasticsearch
input
{
kafka {
bootstrap_servers => “localhost:9092”
topics => [“app-logs”]
}
}
filter
{
json {
source => “message”
}
}
output {
elasticsearch {
hosts => [“http://localhost:9200"]
index => “app-logs-%{+YYYY.MM.dd}”
}
Get Puja Poddar’s stories in your inbox
Join Medium for free to get updates from this writer.
}
Logstash Configuration Example (Kafka to Elasticsearch)
input {
kafka {
bootstrap_servers => “localhost:9092”
topics => [“maxwell-mydb.users”]
codec => “json”
group_id => “logstash-group”
}
}
filter {
if [type] == “insert” or [type] == “update” {
mutate {
remove_field => [“@version”, “host”, “path”]
}
} else {
drop {} # optional: drop delete events
}
}
output {
elasticsearch {
hosts => [“http://localhost:9200"]
index => “users-cdc-%{+YYYY.MM.dd}”
document_id => “%{primary_key_id}” # Optional: use a unique ID for updates
}
stdout { codec => rubydebug }
}
Notes:
- Make sure Maxwell is publishing to Kafka in JSON format (default).
- Replace primary_key_id with your actual key from the Maxwell JSON.
