当前位置：首页 > news >正文

使用Logstash实现PostgreSQL到Elasticsearch的数据摄取

news 2026/6/14 5:51:08

使用Logstash实现PostgreSQL到Elasticsearch的数据摄取

什么是Logstash？

Logstash是Elastic提供的开源数据处理管道工具，用于摄取、转换和将数据发送到不同源，包括Elasticsearch、Kafka、平面文件等。

Logstash管道包含三个不同的处理过程：

输入：从中收集数据以进行摄取的数据源
过滤器：使用Grok、Mutate、Date等插件转换（清理、聚合等）数据
输出：摄取的目标（Elasticsearch、平面文件、数据库等）

以下是使用Logstash将数据发送到Elasticsearch的先决条件：

系统上安装了Logstash和Postgres的JDBC驱动程序
具有要同步的表或函数的Postgres数据库
正在运行的Elasticsearch实例

Logstash设置（Windows版）

以下是本地安装和运行Logstash的简要步骤。

1. 安装Java

从官方Oracle网站下载JDK包（Java 8或更高版本）。下载完成后，将文件解压缩到首选位置。

解压缩文件后，需要添加环境变量以便系统识别Java命令。

转到环境变量，添加一个名为JAVA_HOME的新变量，并将其指向Java文件所在的目录。将%JAVA_HOME%\bin附加到路径中。

要验证安装是否成功，请转到命令提示符并运行以下命令：

java -version

如果一切设置正确，它将显示Java版本。

2. 安装Logstash

要安装Logstash，请从官方Elastic网站下载包，并将其解压缩到首选位置。

要在本地测试，请打开命令提示符，导航到Logstash文件夹中的bin文件夹，并运行以下命令：

logstash -e "input { stdin {} } output { stdout {} }"

Logstash摄取管道

1. 安装所需的JDBC驱动程序

从官方PostgreSQL网站下载Postgres驱动程序。将jar文件放在可访问的位置。

2. 创建Logstash管道

以下是示例管道：

input {jdbc {jdbc_driver_library => "c:/logstash/jdbc/postgresql.jar"jdbc_driver_class => "org.postgresql.Driver"jdbc_connection_string => "${JDBC_HOST}"jdbc_user => "${DB_USER}"jdbc_password => "${DB_PWD}"jdbc_paging_enabled => truejdbc_page_size => 1000schedule => "* * * * *"  # 计划每分钟运行一次statement => "SELECT * FROM employee WHERE updated_at > :sql_last_value"use_column_value => truetracking_column => "updated_at"tracking_column_type => "timestamp"last_run_metadata_path => "c:/logstash/employee.tracker"}
}filter {mutate {remove_field => ["date", "@timestamp", "host"]}# 如果需要解析JSON字段的示例json {source => "first_name"target => "name"}
}output {stdout { codec => json_lines }elasticsearch {hosts => ["http://localhost:9200"]index => "my_table_index"custom_headers => {"Authorization" => "${AUTH_KEY}"}document_id => "%{table_id}" # 表中的唯一标识符timeout => 120}
}

上述管道用于增量摄取。这意味着它会跟踪最后一次运行，并从最后一次运行开始获取记录，按照计划摄取数据。

以下是使用的关键概念：

输入：

jdbc_driver_library - JDBC驱动程序文件(.jar)的存储位置
jdbc_driver_class - 正在使用的驱动程序类
jdbc_connection_string - postgres数据库连接字符串
jdbc_user - 数据库用户名
jdbc_password - 用户的数据库密码
paging - 数据将以多页形式发送，每页大小为1000。这将提高管道的性能，并有助于跟踪发送到Elasticsearch的记录数
schedule - 上述管道计划每分钟运行一次。以下是计划的格式：
statement - 管道将执行的SQL语句。要执行复杂的语句，可以将其保存在单独的.sql文件中，并将文件路径提及到statement_filepath而不是statement。最好使用视图或物化视图而不是具有复杂连接的查询。

最后一部分用于增量摄取：

use_column_value => true
tracking_column => "updated_dt"
tracking_column_type => "timestamp"
last_run_metadata_path => "c:/project/logstash/date.tracker"

use_column_value设置为true。它让Logstash知道跟踪在tracking_column中使用的列updated_at的实际值，而不是使用上次运行查询的时间。在这种情况下，:sql_last_value将使用updated_dt值。
如果设置为false，Logstash将使用上次查询执行时间作为:sql_last_value。
最后一次运行时间将保存在last_run_metadata_path中提到的文件中。它将用于跟踪管道最后一次运行的时间。

过滤器
这是一个可选部分，用于在将数据发送到目标之前操作数据。

在上述管道中，日期字段正在从摄取中删除。此外，它还将数据中的first_name发送到目标中的name字段。

输出
此部分定义数据的目标。在这种情况下，它是Elasticsearch端点、授权密钥（如果有）、elastic索引、document_id。document_id是索引中elastic文档的唯一标识符。如果未提及此字段，Elasticsearch将自动为文档分配唯一标识符。

在增量摄取的情况下，建议定义此字段。在摄取期间，Elasticsearch将在索引中查找此字段；如果匹配，它将更新同一文档。

如果未定义该字段，它将在索引中创建一个新文档，从而导致重复记录。

运行管道

要运行此管道，请打开命令提示符，转到Logstash文件夹，并运行以下命令：

bin/logstash -f c:/logstash/sample_pipeline.conf

以下是管道的输出。

来自Elasticsearch索引的输出。

{"took": 1,"timed_out": false,"_shards": {"total": 1,"successful": 1,"skipped": 0,"failed": 0},"hits": {"total": {"value": 3,"relation": "eq"},"max_score": 1.0,"hits": [{"_index": "testing","_id": "1","_score": 1.0,"_source": {"name": "James","id": 1,"last_name": "Smith","updated_dt": "2024-12-12T16:10:57.349Z","@version": "1","@timestamp": "2025-06-25T20:41:02.167442600Z"}},{"_index": "testing","_id": "2","_score": 1.0,"_source": {"name": "John","id": 2,"last_name": "Doe","updated_dt": "2024-12-12T16:10:57.349Z","@version": "1","@timestamp": "2025-06-25T20:41:02.169021400Z"}},{"_index": "testing","_id": "3","_score": 1.0,"_source": {"name": "Kate","id": 3,"last_name": "Williams","updated_dt": "2024-12-12T16:10:57.349Z","@version": "1","@timestamp": "2025-06-25T20:41:02.170098800Z"}}]}
}

这种方法有几个优点：