PG数据库同步数据到ES数据库

思想

ES数据库同步PG数据库在本文主要用到的思想是：在PG数据库的数据录入以及更新时，如下图所示，会有其对应的字段modify_time记录最后的修改时间。程序会记录同步到ES数据库中最后一条数据的修改时间，利用线程间隔10s检查一次PG数据库是否有modify_time > 记录的最后修改时间。如果有，将最新的数据同步到ES数据库，并修改记录时间。如果没有，继续每隔10s检查一次。

最终的实现效果：

程序实现的主要代码

def update_pg2es():global last_modify_timeprint(type(last_modify_time), last_modify_time)print("检测PG数据库是否有数据更新...")#连接数据库pg_conn = psycopg2.connect(config.CONNECT_PG_DB_URL)pg_cursor = pg_conn.cursor()select_sql = "select * from test where modify_time > '{}' order by modify_time".format(last_modify_time)pg_cursor.execute(select_sql)datas = pg_cursor.fetchall()# 关闭连接pg_conn.closepg_cursor.closees_data_list = []for row in datas:# 获取当前行数据的列与值得字典column_values = get_row_colum_info(pg_cursor, row)es_data = translate_entity_date_to_es(column_values)es_data_list.append(es_data)last_modify_time = column_values['modify_time']index_entity = config.entity_indextype_entity = config.entity_typenum = 10000  # 批量存入的个数while (len(es_data_list) / num >= 0):if (math.floor(len(es_data_list) / num) == 0):  # 最后了es_result = es_service.insert_data_list(index_entity, type_entity, es_data_list[:])  # 取剩下的print("存入{}个数据到ES".format(len(es_data_list)))breakprint("存入{}个数据到ES".format(num))print("前五个数据：")print(index_entity, type_entity, es_data_list[:5])es_result = es_service.insert_data_list(index_entity, type_entity, es_data_list[:num])  # 存储一定的数量es_data_list = es_data_list[num:]  # 取剩下的if "ok" != es_result:print("存储实体数据到es中失败")break  # 如果某一次存入有问题，直接退出# 线程定时执行t = threading.Timer(10, update_pg2es)t.start()if __name__ == '__main__':update_pg2es()

注意:

select now()::timestamp(6)without time zone SQL语句可以查询当前时间（不带时区，精度为秒后面保存6位）
select_sql = "select * from test where modify_time > '{}' order by modify_time".format(last_modify_time)将sql语句赋值给select_sql，sql语句的意思为选择出修改时间大于记录时间的数据，并将数据按照修改时间进行排序。此处注意记录时间两边需要加单引号才能保证sql语句的正常运行。

PG数据库同步数据到ES数据库相关推荐

mysql数据库同步数据至oracle数据库
一.同步方法使用Navicat Premium工具进行数据同步,首先分别建立数据库: ss_sysright.ss_safe数据传输配置文件. 然后使用自动运行,创建自动运行计划任务. 二.操作步骤 ...
sql 不同数据库同步数据_什么是SQL数据同步
sql 不同数据库同步数据 SQL Data Sync is a service that allows synchronizing data across multiple Azure SQL da ...
用Aliyun E-MapReduce集群的sqoop工具和数据库同步数据如何配置网络
如果您的E-MapReduce集群需要和集群之外的数据库同步数据,需要确保网络是联通的.本文就RDS,ecs自搭,云下私有数据库三种情况,分别介绍如何配置网络. 一.RDS 经典网络RDS 想要访问经 ...
ubuntun系统mysql数据库同步_Canal 实现 Mysql数据库实时数据同步
简介 1.1 canal介绍 Canal是一个基于MySQL二进制日志的高性能数据同步系统.Canal广泛用于阿里巴巴集团(包括https://www.taobao.com),以提供可靠的低延迟增 ...
oracle同步数据adg_[adg数据库同步机制]三分钟读懂Oracle数据库容灾架之DataGuard
在线QQ客服:1922638 专业的SQL Server.MySQL数据库同步软件 Oracle数据库目前依然处于商用数据库的霸主地位. 运行在Oracle数据库上的核心业务及核心数据的安全性尤为重要 ...
python 数据库同步_python实现不同数据库间数据同步功能
功能描述数据库间数据同步方式很多,在上篇博文中有总结.本文是用py程序实现数据同步. a数据库中有几十张表,要汇聚到b数据库中,且表结构一致,需要准实时的进行数据同步,用工具实现时对其控制有限且配置 ...
IFIX往Mysql数据库同步数据方法
一.操作环境 Windows server12,ifix数据库,Mysql数据库(5.7) 二.操作步骤 1.在Windows Server12上安装Mysql ODBC(官网下载即可) 由于Wind ...
MySQL数据库同步数据到Oracle
通过一个工具可以将不同数据库的数据同步到另一个数据库中,以MySQL数据导入到Oracle为例: 导入工具 Navicat Premium 15 工具下载地址 Navicat | 产品操作步骤: 1 ...
logstash同步数据到es
1.下载下载地址 https://www.elastic.co/cn/downloads/logstash 2.安装解压即可 3.配置有两种同步,一种是增量同步,增量同步则只同步两个数据库不同的 ...

PG数据库同步数据到ES数据库

思想

最终的实现效果：

程序实现的主要代码

PG数据库同步数据到ES数据库相关推荐

最新文章

热门文章