服务器之家：专注于服务器技术及软件下载分享
分类导航

Mysql|Sql Server|Oracle|Redis|MongoDB|PostgreSQL|Sqlite|DB2|mariadb|Access|数据库技术|

服务器之家 - 数据库 - PostgreSQL - PostgreSQL 数据同步到ES 搭建操作

PostgreSQL 数据同步到ES 搭建操作

2021-03-11 21:14将臣三代 PostgreSQL

这篇文章主要介绍了PostgreSQL 数据同步到ES 搭建操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧

安装python 和dev 开发包

?
1
2
3
4
5
[root@rtm2 Packages]# rpm -ivh python-devel-2.7.5-58.el7.x86_64.rpm
准备中...       ################################# [100%]
正在升级/安装...
 1:python-devel-2.7.5-58.el7  ################################# [100%]
[root@rtm2 Packages]# ls

安装 multicorn

?
1
2
3
4
5
6
7
8
9
10
11
12
[root@rtm2 multicorn-1.3.5]# make
Python version is 2.7
gcc -Wall -Wmissing-prototypes -Wpointer-arith -Wdeclaration-after-statement -Wendif-labels -Wmissing-format-attribute -Wformat-security -fno-strict-aliasing -fwrapv -fexcess-precision=standard -O2 -fPIC -I/usr/include/python2.7 -I/usr/include/python2.7 -I. -I./ -I/opt/pgsql-10/include/server -I/opt/pgsql-10/include/internal -D_GNU_SOURCE -c -o src/errors.o src/errors.c
gcc -Wall -Wmissing-prototypes -Wpointer-arith -Wdeclaration-after-statement -Wendif-labels -Wmissing-format-attribute -Wformat-security -fno-strict-aliasing -fwrapv -fexcess-precision=standard -O2 -fPIC -I/usr/include/python2.7 -I/usr/include/python2.7 -I. -I./ -I/opt/pgsql-10/include/server -I/opt/pgsql-10/include/internal -D_GNU_SOURCE -c -o src/python.o src/python.c
gcc -Wall -Wmissing-prototypes -Wpointer-arith -Wdeclaration-after-statement -Wendif-labels -Wmissing-format-attribute -Wformat-security -fno-strict-aliasing -fwrapv -fexcess-precision=standard -O2 -fPIC -I/usr/include/python2.7 -I/usr/include/python2.7 -I. -I./ -I/opt/pgsql-10/include/server -I/opt/pgsql-10/include/internal -D_GNU_SOURCE -c -o src/query.o src/query.c
gcc -Wall -Wmissing-prototypes -Wpointer-arith -Wdeclaration-after-statement -Wendif-labels -Wmissing-format-attribute -Wformat-security -fno-strict-aliasing -fwrapv -fexcess-precision=standard -O2 -fPIC -I/usr/include/python2.7 -I/usr/include/python2.7 -I. -I./ -I/opt/pgsql-10/include/server -I/opt/pgsql-10/include/internal -D_GNU_SOURCE -c -o src/multicorn.o src/multicorn.c
gcc -Wall -Wmissing-prototypes -Wpointer-arith -Wdeclaration-after-statement -Wendif-labels -Wmissing-format-attribute -Wformat-security -fno-strict-aliasing -fwrapv -fexcess-precision=standard -O2 -fPIC -shared -o multicorn.so src/errors.o src/python.o src/query.o src/multicorn.o -L/opt/pgsql-10/lib -Wl,--as-needed -Wl,-rpath,'/opt/pgsql-10/lib',--enable-new-dtags -lpthread -ldl -lutil -lm -lpython2.7 -lpthread -ldl -lutil -lm -lpython2.7 -Xlinker -export-dynamic
.//preflight-check.sh
cp sql/multicorn.sql sql/multicorn--1.3.5.sql
[root@rtm2 multicorn-1.3.5]# make install
Python version is 2.7
...

安装pg-es-fdw-master

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
[root@rtm2 multicorn-1.3.5]# cd ../pg-es-fdw-master
[root@rtm2 pg-es-fdw-master]# ls
demo.sh dite LICENSE README.md setup.py
[root@rtm2 pg-es-fdw-master]# python setup.py build
running build
running build_py
creating build
creating build/lib
creating build/lib/dite
copying dite/__init__.py -> build/lib/dite
[root@rtm2 pg-es-fdw-master]# python setup.py install
running install
running bdist_egg
running egg_info
creating dite.egg-info
writing dite.egg-info/PKG-INFO

安装插件 multicorn

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
[postgres@rtm2 ~]$ psql
psql (10.3)
Type "help" for help.
postgres=# select * from pg_extension;
 extname | extowner | extnamespace | extrelocatable | extversion | extconfig | extcondition
---------+----------+--------------+----------------+------------+-----------+--------------
 plpgsql |  10 |   11 | f    | 1.0  |   |
(1 row)
postgres=# CREATE EXTENSION multicorn;
CREATE EXTENSION
postgres=# psql
postgres=# select * from pg_extension;
 extname | extowner | extnamespace | extrelocatable | extversion | extconfig | extcondition
-----------+----------+--------------+----------------+------------+-----------+--------------
 plpgsql |  10 |   11 | f    | 1.0  |   |
 multicorn |  10 |   2200 | t    | 1.3.5  |   |
(2 rows)
postgres=# CREATE SERVER multicorn_es FOREIGN DATA WRAPPER multicorn OPTIONS(wrapper 'dite.ElasticsearchFDW');
CREATE SERVER
postgres=#

es

?
1
2
3
4
[root@rtm2 config]# vi elasticsearch.yml
node.name: "es-node1"
network.host: 192.168.31.121
discovery.zen.ping.unicast.hosts: ["192.168.31.121"]
?
1
2
3
4
5
6
7
8
9
10
[root@rtm2 config]# vi /etc/sysctl.conf
vm.max_map_count=262144
sysctl -p
[root@rtm2 config]# vi /etc/security/limits.conf
# End of file
 root soft nofile 65536
root hard nofile 65536
root soft nproc 4096
root hard nproc 4096
~

启动es

?
1
2
3
4
[root@rtm2 bin]# ls
elasticsearch  elasticsearch.in.bat elasticsearch-service-mgr.exe elasticsearch-service-x86.exe plugin.bat
elasticsearch.bat elasticsearch.in.sh elasticsearch-service-x64.exe plugin       service.bat
[root@rtm2 bin]# ./bin/elasticsearch
?
1
2
3
4
test=# CREATE FOREIGN TABLE pp_es (id bigint,age bigint) SERVER multicorn_es OPTIONS (host
test(# '192.168.31.121', port '9200', node 'es-node1', index 'pp');
CREATE FOREIGN TABLE
test=#

创建触发器和外部表

?
1
2
3
4
5
6
7
8
9
10
11
test=# CREATE OR REPLACE FUNCTION index_pp() RETURNS trigger AS $def$
test$# BEGIN
test$# INSERT INTO pp_es (id, age) VALUES
test$# (NEW.id, NEW.age);
test$# RETURN NEW;
test$# END;
test$# $def$ LANGUAGE plpgsql;
CREATE FUNCTION
test=# CREATE TRIGGER es_insert_pp AFTER INSERT ON pp FOR EACH ROW EXECUTE PROCEDURE index_pp();
CREATE TRIGGER
test=#

新增数据测试

?
1
2
3
4
5
6
7
8
test=# insert into pp (id,age) values (1,11);
INSERT 0 1
test=# select * from pp;
 id | age
----+-----
 1 | 11
(1 row)
test=#

检查es数据

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
[root@rtm2 ~]# curl 'http://192.168.31.121:9200/es-node1/_search?q=*:*&pretty'
{
 "took" : 104,
 "timed_out" : false,
 "_shards" : {
 "total" : 5,
 "successful" : 5,
 "failed" : 0
 },
 "hits" : {
 "total" : 2,
 "max_score" : 1.0,
 "hits" : [ {
  "_index" : "es-node1",
  "_type" : "pp",
  "_id" : "1",
  "_score" : 1.0,
  "_source":{"age": "11"}
 }, {
  "_index" : "es-node1",
  "_type" : "pp",
  "_id" : "2",
  "_score" : 1.0,
  "_source":{"age": "22"}
 } ]
 }
}
[root@rtm2 ~]#

创建更新触发器

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
test=# CREATE OR REPLACE FUNCTION updadeIndex_pp() RETURNS trigger AS $def$
BEGIN
UPDATE pp_es SET
id = NEW.id,
age = NEW.age
where id =NEW.id;
RETURN NEW;
END;
$def$ LANGUAGE plpgsql;
CREATE FUNCTION
test=# ^C
test=#
test=# CREATE TRIGGER es_update_pp AFTER UPDATE OF id, age ON pp FOR EACH ROW WHEN (OLD.* IS DISTINCT
test(# FROM NEW.*)EXECUTE PROCEDURE updadeIndex_pp();
CREATE TRIGGER
test=#

更新表数据

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
test=# select * from pp;
 id | age
----+-----
 1 | 11
 2 | 22
 3 | 22
(3 rows)
test=# update pp a set a.age = 33 where a.id = 3;
ERROR: column "a" of relation "pp" does not exist
LINE 1: update pp a set a.age = 33 where a.id = 3;
      ^
test=# update pp set age = 33 where id = 3;
UPDATE 1
test=# select * from pp;
 id | age
----+-----
 1 | 11
 2 | 22
 3 | 33
(3 rows)
test=#

es查询变更

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
[root@rtm2 ~]# curl 'http://192.168.31.121:9200/es-node1/_search?q=*:*&pretty'
{
 "took" : 4,
 "timed_out" : false,
 "_shards" : {
 "total" : 5,
 "successful" : 5,
 "failed" : 0
 },
 "hits" : {
 "total" : 3,
 "max_score" : 1.0,
 "hits" : [ {
  "_index" : "es-node1",
  "_type" : "pp",
  "_id" : "1",
  "_score" : 1.0,
  "_source":{"age": "11"}
 }, {
  "_index" : "es-node1",
  "_type" : "pp",
  "_id" : "2",
  "_score" : 1.0,
  "_source":{"age": "22"}
 }, {
  "_index" : "es-node1",
  "_type" : "pp",
  "_id" : "3",
  "_score" : 1.0,
  "_source":{"age": "33"}
 } ]
 }
}
[root@rtm2 ~]#

补充：logstash同步pgsql数据到Elasticsearch

一、对于logstash的配置我就不在多说，主要是三部分，input、filter、output的配置

二、配置步骤

1、input配置

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
input {
 stdin {
 }
 jdbc {
  jdbc_connection_string => "jdbc:postgresql://127.0.0.1:5432/world"
  jdbc_user => "postgres"
  jdbc_password => "zhang123"
  jdbc_driver_library => "D:\logstash-6.4.0\bin\pgsql\postgresql-42.2.5.jar"
  jdbc_driver_class => "org.postgresql.Driver"
  jdbc_paging_enabled => "true"
  jdbc_page_size => "300000"
  use_column_value => "true"
  tracking_column => "id"
  statement_filepath => "D:\logstash-6.4.0\bin\pgsql\jdbc.sql"
 schedule => "* * * * *"
 type => "jdbc"
 jdbc_default_timezone =>"Asia/Shanghai"
 }
}

2、filter配置

?
1
2
3
4
5
6
filter {
 json {
  source => "message"
  remove_field => ["message"]
 }
}

3、output 配置，就是elasticsearch的基本配置

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
output {
 elasticsearch {
  hosts => ["localhost:9200"]
  index => "test_out"
 template => "D:\logstash-6.4.0\bin\pgsql\es-template.json"
 template_name => "t-statistic-out-logstash"
 template_overwrite => true
 document_type => "out"
  document_id => "%{id}"
 }
 stdout {
  codec => json_lines
 }
}

以上就是整个logstash 的jdbc.conf

4、es-template.json的配置

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
{
 "template" : "t-statistis-out-template",
 "order":1,
 "settings": {
   "index": {
    "refresh_interval": "5s"
   }
  },
 "mappings": {
   "_default_": {
 "_all" : {"enabled":false},
    "dynamic_templates": [
     {
    "message_field" : {
    "match" : "message",
    "match_mapping_type" : "string",
    "mapping" : { "type" : "string", "index" : "not_analyzed" }
    }
   }, {
    "string_fields" : {
    "match" : "*",
    "match_mapping_type" : "string",
    "mapping" : { "type" : "string", "index" : "not_analyzed" }
    }
   }
    ],
    "properties": {
     "@timestamp": {
      "type": "date"
     },
     "@version": {
      "type": "keyword"
     },    
  "id": {
      "type": "keyword"
     },
  "name": {
      "type": "keyword"
     },
  "pp": {
      "type": "keyword"
     } 
    }
   }
  },
  "aliases": {}
 
}

最后就是就是下载好pgsql的连接驱动，这个官网可以下载；配置好自己的数据库表格的数据

启动命令：进入到logstash的bin目录下，自己的logstash配置都是放在bin的pgsql这个目录下面（这个自己随意创建位置都可以）

?
1
logstash.bat -f ./pgsql/jdbc.conf

以上为个人经验，希望能给大家一个参考，也希望大家多多支持服务器之家。如有错误或未考虑完全的地方，望不吝赐教。

原文链接：https://blog.csdn.net/yaoqiancuo3276/article/details/80612090

  • PostgreSQL
  • ES
  • 数据同步

    • 延伸 · 阅读

    精彩推荐
    • PostgreSQLpostgresql 中的to_char()常用操作

      postgresql 中的to_char()常用操作

      这篇文章主要介绍了postgresql 中的to_char()常用操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧...

      J符离13432021-04-12
    • PostgreSQLPostgresql查询效率计算初探

      Postgresql查询效率计算初探

      这篇文章主要给大家介绍了关于Postgresql查询效率计算的相关资料，文中通过示例代码介绍的非常详细，对大家学习或者使用Postgresql具有一定的参考学习价...

      轨迹4622020-05-03
    • PostgreSQLpostgresql 数据库中的数据转换

      postgresql 数据库中的数据转换

      postgres8.3以后，字段数据之间的默认转换取消了。如果需要进行数据变换的话，在postgresql数据库中，我们可以用"::"来进行字段数据的类型转换。...

      postgresql教程网12482021-10-08
    • PostgreSQLRDS PostgreSQL一键大版本升级技术解密

      RDS PostgreSQL一键大版本升级技术解密

      一、PostgreSQL行业位置 （一）行业位置 在讨论PostgreSQL（下面简称为PG）在整个数据库行业的位置之前，我们先看一下阿里云数据库在全球的数据库行业里的...

      未知1192023-05-07
    • PostgreSQL分布式 PostgreSQL之Citus 架构

      分布式 PostgreSQL之Citus 架构

      节点 Citus 是一种 PostgreSQL 扩展，它允许数据库服务器(称为节点)在“无共享(shared nothing)”架构中相互协调。这些节点形成一个集群，允许 PostgreSQL 保存比单...

      未知802023-05-07
    • PostgreSQLPostgreSQL标准建表语句分享

      PostgreSQL标准建表语句分享

      这篇文章主要介绍了PostgreSQL标准建表语句分享，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧...

      码上得天下7962021-02-27
    • PostgreSQL深入理解PostgreSQL的MVCC并发处理方式

      深入理解PostgreSQL的MVCC并发处理方式

      这篇文章主要介绍了深入理解PostgreSQL的MVCC并发处理方式,文中同时介绍了MVCC的缺点,需要的朋友可以参考下 ...

      PostgreSQL教程网3622020-04-25
    • PostgreSQLPostgresql开启远程访问的步骤全纪录

      Postgresql开启远程访问的步骤全纪录

      postgre一般默认为本地连接，不支持远程访问，所以如果要开启远程访问，需要更改安装文件的配置。下面这篇文章主要给大家介绍了关于Postgresql开启远程...

      我勒个去6812020-04-30
    最近更新
    编辑推荐
    阅读排行
    热门标签
    5432 　 postgresql命令 　 数据库教程 　 入门命令 　 死锁进程 　 PostgreSQL教程 　 表继承 　 事物隔离 　 性能提升技巧 　 PL/pgSQL 　 过程语言 　 partitioning 　 PostgreSQL安装教程 　 跨库访问 　 postgresql数据库 　 并发处理 　 MVCC 　 Citus 　 OID 　 XID 　 表空间大小 　 Postgre 　 phpPgAdmin 　 phpPgAdmin错误 　 phpPgAdmin问题 　 phpPgAdmin配置文件 　 app应用 　 PostgreSQL配置 　 Postgresql下载 　 PostgreSQL入门教程 　
    © 2019-2023 服务器之家(www.zzvips.com) 版权所有关于我们联系我们版权申明网站地图