09月27, 2019

Postgresql 同步 ElasticSearch 环境部署

安装环境准备

  • Ubuntbu 18.04LTS
  • 数据库:PostgreSQL 10
  • 数据库客户端:pgAdmin 4
  • Python 2.7.10

Docker 部署 ElasticSearch

1) 拉取镜像

docker pull docker.elastic.co/elasticsearch/elasticsearch:7.1.1

2) 运行容器

ElasticSearch的默认端口是9200,我们把宿主环境9200端口映射到Docker容器中的9200端口,就可以访问到Docker容器中的ElasticSearch服务了,同时我们把这个容器命名为es。

docker run -d --name es -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" docker.elastic.co/elasticsearch/elasticsearch:7.1.1

3) 配置跨域

# 显示文件
ls
结果如下:
LICENSE.txt  README.textile  config  lib   modules
NOTICE.txt   bin             data    logs  plugins

# 进入配置文件夹
cd config

# 显示文件
ls
结果如下:
elasticsearch.keystore  ingest-geoip  log4j2.properties  roles.yml  users_roles
elasticsearch.yml       jvm.options   role_mapping.yml   users

# 修改配置文件
vi elasticsearch.yml

# 加入跨域配置
http.cors.enabled: true
http.cors.allow-origin: "*"

4) 重启容器

docker restart es

安装 PostgreSQL 插件 multicorn

github下载 Multicorn:

git clone https://github.com/Kozea/Multicorn /tmp/multicorn
cd $_

make
make install

如遇安装错误,需要安装python dev库

fatal error: Python.h: No such file or directory

Ubuntu下 fatal error: Python.h: No such file or directory 解决方法 根据使用的Python版本安装python-dev库。

sudo apt-get install python-dev   # for python2.x installs
sudo apt-get install python3-dev  # for python3.x installs
sudo apt-get install python3.6-dev
sudo apt-get install python-setuptools

pg 数据库中新建一个 EXTENSION

create extension multicorn ;

安装 PostgreSQL 插件 pg-es-fdw

git clone https://github.com/Mikulas/pg-es-fdw /tmp/pg-es-fdw
cd $_
sudo python setup.py install
service postgresql restart

基于 multicorn 创建 foreign server

CREATE SERVER multicorn_es FOREIGN DATA WRAPPER multicorn
OPTIONS (
  wrapper 'dite.ElasticsearchFDW'
);

PostgreSQL 的建 server 的语法不再赘述。wrapper 的值 dite.ElasticsearchFDW 的定义就在 pg-es-fdw 插件的 dite 目录 init.py 文件里,有兴趣的可以看看。

创建测试表

CREATE TABLE articles (
    id serial PRIMARY KEY,
    title text NOT NULL,
    content text NOT NULL,
    created_at timestamp
);

创建外部表

CREATE FOREIGN TABLE articles_es (
    id bigint,
    title text,
    content text
) SERVER multicorn_es OPTIONS (host '127.0.0.1', port '9200', node 'test', index 'articles');

创建触发器

对实体表,创建触发器函数,在用户对实体表插入,删除,更新时,通过触发器函数自动将数据同步到对应ES的外部表。同步过程调用 FDW 的接口,对 ES 进行索引的建立,更新,删除:

CREATE OR REPLACE FUNCTION index_article() RETURNS trigger AS $def$
    BEGIN
        INSERT INTO articles_es (id, title, content) VALUES
            (NEW.id, NEW.title, NEW.content);
        RETURN NEW;
    END;
$def$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION reindex_article() RETURNS trigger AS $def$
    BEGIN
        UPDATE articles_es SET
            title = NEW.title,
            content = NEW.content
        WHERE id = NEW.id;
        RETURN NEW;
    END;
$def$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION delete_article() RETURNS trigger AS $def$
    BEGIN
        DELETE FROM articles_es a WHERE a.id = OLD.id;
        RETURN OLD;
    END;
$def$ LANGUAGE plpgsql;

CREATE TRIGGER es_insert_article
    AFTER INSERT ON articles
    FOR EACH ROW EXECUTE PROCEDURE index_article();

CREATE TRIGGER es_update_article
    AFTER UPDATE OF title, content ON articles
    FOR EACH ROW
    WHEN (OLD.* IS DISTINCT FROM NEW.*)
    EXECUTE PROCEDURE reindex_article();

CREATE TRIGGER es_delete_article
    BEFORE DELETE ON articles
    FOR EACH ROW EXECUTE PROCEDURE delete_article();

测试验证

插入一条表记录:

insert into articles(title, content) values ('测试内容1', '测试标题1');
insert into articles(title, content) values ('测试内容2', '测试标题2');

查询 ES ,检查数据是否已同步:

curl 'localhost:9200/test/articles/_search?q=*:*&pretty'

结果如下:

{
  "took" : 4,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 2,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "test",
        "_type" : "articles",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "content" : "测试标题1",
          "title" : "测试内容1"
        }
      },
      {
        "_index" : "test",
        "_type" : "articles",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "content" : "测试标题2",
          "title" : "测试内容2"
        }
      }
    ]
  }
}

参考资料

本文链接:http://57km.cc/post/deploy ElasticSearch using docker.html

-- EOF --

Comments