POSTGRESQL中ETL、fdw的平行替换

2024-01-13 23:36
文章标签 postgresql etl 替换 平行 fdw

本文主要是介绍POSTGRESQL中ETL、fdw的平行替换，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

POSTGRESQL中ETL、fdw的平行替换

01、简介

“ 在我前两次的文章中，说到postgresql对于python的支持，其实很多功能也就可以封装进入的postgresql数据库中去。比如fdw、etl等，本文将以此为叙述点，进行演示展示”

d9de783ffe5e4224852626e3f2388609_0.png

在postgresql数据库中fdw的支持，在创建和使用上都不上太方便，特别是fdw在用表级别关联的时候，性能会大大折扣，因为fdw的数据并不会落地到本地​。所以我们可以利用postgresql对于python的支持，自行封装一个库对库的调度工具，将远端数据进行落地​后再次使用。对于使用的便利性，读者可自行​对比。

02、postgresql16.1的安装

安装依赖

yum install -y bison flex readline-devel zlib-devel zlib zlib-devel gcc  gcc-c++ openssl-devel python3  python3-devel libicu-devel ncurses-devel sqlite-devel tk-devel gcc make

添加用户

useradd postgres 
vim /etc/sudo

在101行以下添加以下内容


postgres ALL=(ALL)     NOPASSWD: ALL

进入官网找到链接，这里使用源码安装。

wget https://ftp.postgresql.org/pub/source/v16.1/postgresql-16.1.tar.gz

解压并进入解压目录

mv postgresql-16.1.tar.gz /home/postgressu - postgres tar -zxf postgresql-16.1.tar.gzcd postgresql-16.1

这里编译python支持还是很重要。–with-python 自行构建plpython3u插件


./configure --prefix=/home/postgres/pg --with-openssl  --with-pythonmake && make install

编辑环境变量


cd 
vim .bash_profile

加入以下环境变量

export PATH=/home/postgres/pg/bin:$PATH 
export PGDATA=/home/postgres/pg/data

加载环境变量


source ~/.bash_profile

初始化数据库


initdb -D $PGDATA -U postgres -W 
(输入超级用户密码两次)
pg_ctl start 
pg_ctl status

进入数据库创建拓展


CREATE EXTENSION plpython3u CASCADE;

02、创建支持跨库访问的函数

首先下载python链接数据库所需module

postgres=# \! pip3 install -i https://mirrors.aliyun.com/pypi/simple/ cx_Oracle pyodbc pymysql --user 
Looking in indexes: https://mirrors.aliyun.com/pypi/simple/
Requirement already satisfied: cx_Oracle in ./.local/lib/python3.6/site-packages (8.3.0)
Collecting pyodbcDownloading https://mirrors.aliyun.com/pypi/packages/27/5c/5e472d714dea2a634bd79df6b8ace55737a9f50c8fbb3b15521fceda4694/pyodbc-4.0.39-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (330 kB)|████████████████████████████████| 330 kB 2.8 MB/s            
Collecting pymysqlDownloading https://mirrors.aliyun.com/pypi/packages/4f/52/a115fe175028b058df353c5a3d5290b71514a83f67078a6482cff24d6137/PyMySQL-1.0.2-py3-none-any.whl (43 kB)|████████████████████████████████| 43 kB 2.4 MB/s             
Installing collected packages: pyodbc, pymysql
Successfully installed pymysql-1.0.2 pyodbc-4.0.39

在链接远程Oracle数据库，需要下载指定的客户端，本文使用的是oracle 19C

wget https://download.oracle.com/otn_software/linux/instantclient/1921000/oracle-instantclient19.21-basic-19.21.0.0.0-1.x86_64.rpm
sudo rpm -ivh oracle-instantclient19.21-basic-19.21.0.0.0-1.x86_64.rpm

编辑环境变量

vim /etc/profile

配置以下环境变量值

export LD_LIBRARY_PATH=/usr/lib/oracle/19.21/client64/lib:$LD_LIBRARY_PATH

加载环境变量

source /etc/profile

在postgresql数据库中创建具有跨库链接mysql\oracle\sqlserver功能的function。

CREATE OR REPLACE FUNCTION fdw_db(db_type varchar(100),host VARCHAR(100),port integer, username VARCHAR(100), password VARCHAR(100), db_name VARCHAR(100),tablename varchar(100))
RETURNS text AS $$import cx_Oracle
import pyodbc
import pymysqldef read_data_from_database(db_type, host, port, username, password, db_name, table_name):result_values = []  # Initialize as an empty list# 读取Oracle数据库中指定表数据的函数if db_type.lower() == 'oracle':connection_string = f"{username}/{password}@{host}:{port}/{db_name}"connection = cx_Oracle.connect(connection_string)cursor = connection.cursor()cursor.execute(f'SELECT * FROM {table_name}')result = cursor.fetchall()cursor.close()connection.close()# 将结果转换为支持INSERT INTO的VALUES语句for row in result:values_str = ', '.join([f"'{value}'" if isinstance(value, str) else str(value) for value in row])result_values.append(f'({values_str})')# 读取SQL Server数据库中指定表数据的函数elif db_type.lower() == 'sqlserver':connection = pyodbc.connect(f"DRIVER={{SQL Server}};SERVER={host};port={port};DATABASE={db_name};UID={username};PWD={password}")cursor = connection.cursor()cursor.execute(f'SELECT * FROM {table_name}')result = cursor.fetchall()cursor.close()connection.close()# 将结果转换为支持INSERT INTO的VALUES语句for row in result:values_str = ', '.join([f"'{value}'" if isinstance(value, str) else str(value) for value in row])result_values.append(f'({values_str})')# 读取MySQL数据库中指定表数据的函数elif db_type.lower() == 'mysql':connection = pymysql.connect(host=host, user=username, password=password, database=db_name, port=port)cursor = connection.cursor()cursor.execute(f'SELECT * FROM {table_name}')result = cursor.fetchall()cursor.close()connection.close()# 将结果转换为支持INSERT INTO的VALUES语句for row in result:values_str = ', '.join([f"'{value}'" if isinstance(value, str) else str(value) for value in row])result_values.append(f'({values_str})')else:raise ValueError("Unsupported database type. Supported types: 'oracle', 'sqlserver', 'mysql'")# 返回拼接的VALUES子句return ', '.join(result_values)insert_values = read_data_from_database(db_type, host, port, username, password, db_name, tablename)
return insert_values$$ LANGUAGE plpython3u;

以Oracle作为测试 在Oracle 和PG中均创建测试表conn_fdw
postgresql

-- 创建表 conn_fdw
CREATE TABLE conn_fdw (id integer,name VARCHAR(50),age integer,city VARCHAR(50),salary integer
);

oracle中

-- 创建表 conn_fdw
CREATE TABLE conn_fdw (id NUMBER,name VARCHAR2(50),age NUMBER,city VARCHAR2(50),salary NUMBER
);

Oracle中插入数据

-- 插入20行数据
INSERT INTO conn_fdw VALUES (1, 'John', 30, 'New York', 50000);
INSERT INTO conn_fdw VALUES (2, 'Alice', 25, 'Los Angeles', 60000);
INSERT INTO conn_fdw VALUES (3, 'Bob', 35, 'Chicago', 70000);
INSERT INTO conn_fdw VALUES (4, 'Eva', 28, 'San Francisco', 55000);
INSERT INTO conn_fdw VALUES (5, 'Mike', 32, 'Seattle', 65000);
INSERT INTO conn_fdw VALUES (6, 'Sophia', 29, 'Boston', 75000);
INSERT INTO conn_fdw VALUES (7, 'David', 27, 'Denver', 52000);
INSERT INTO conn_fdw VALUES (8, 'Emily', 31, 'Austin', 68000);
INSERT INTO conn_fdw VALUES (9, 'Daniel', 26, 'Phoenix', 58000);
INSERT INTO conn_fdw VALUES (10, 'Olivia', 33, 'Houston', 72000);
INSERT INTO conn_fdw VALUES (11, 'Liam', 24, 'Portland', 49000);
INSERT INTO conn_fdw VALUES (12, 'Ava', 34, 'Atlanta', 71000);
INSERT INTO conn_fdw VALUES (13, 'Logan', 30, 'Miami', 62000);
INSERT INTO conn_fdw VALUES (14, 'Mia', 28, 'Dallas', 54000);
INSERT INTO conn_fdw VALUES (15, 'Jackson', 29, 'Minneapolis', 67000);
INSERT INTO conn_fdw VALUES (16, 'Sophie', 31, 'Detroit', 59000);
INSERT INTO conn_fdw VALUES (17, 'William', 27, 'Philadelphia', 70000);
INSERT INTO conn_fdw VALUES (18, 'Emma', 32, 'San Diego', 66000);
INSERT INTO conn_fdw VALUES (19, 'James', 26, 'Raleigh', 63000);
INSERT INTO conn_fdw VALUES (20, 'Avery', 35, 'Tampa', 71000);

此时再结合SQL语言进行处理远程连接传过来数据，再创建一个函数用于调用以上创建fdw_db

CREATE OR REPLACE FUNCTION inset_fdw_db(db_type varchar(100),host VARCHAR(100),port integer, username VARCHAR(100), password VARCHAR(100), db_name VARCHAR(100),tablename varchar(100),target_bale varchar(100))
RETURNS void AS $$
declare 
data_values text;
begin 
SELECT   fdw_db(db_type, host, port, username, password, db_name,tablename) into data_values;EXECUTE 'insert into '||target_bale ||' values'||data_values;
end;$$ LANGUAGE plpgsql;

进行调用

 SELECT   inset_fdw_db('oracle', '192.168.48.1', 1521, 'system', 'system', 'orcl', 'CONN_FDW','public.conn_fdw');

进入数据库中查看
此时数据已经落地

postgres=# select *  from CONN_FDW;id | name | age | city | salary 
----+------+-----+------+--------
(0 rows)postgres=#  SELECT   inset_fdw_db('oracle', '192.168.48.1', 1521, 'system', 'system', 'orcl', 'CONN_FDW','public.conn_fdw');inset_fdw_db 
--------------(1 row)postgres=# select *  from CONN_FDW;id |  name   | age |     city      | salary 
----+---------+-----+---------------+--------1 | John    |  30 | New York      |  500002 | Alice   |  25 | Los Angeles   |  600003 | Bob     |  35 | Chicago       |  700004 | Eva     |  28 | San Francisco |  550005 | Mike    |  32 | Seattle       |  650006 | Sophia  |  29 | Boston        |  750007 | David   |  27 | Denver        |  520008 | Emily   |  31 | Austin        |  680009 | Daniel  |  26 | Phoenix       |  5800010 | Olivia  |  33 | Houston       |  7200011 | Liam    |  24 | Portland      |  4900012 | Ava     |  34 | Atlanta       |  7100013 | Logan   |  30 | Miami         |  6200014 | Mia     |  28 | Dallas        |  5400015 | Jackson |  29 | Minneapolis   |  6700016 | Sophie  |  31 | Detroit       |  5900017 | William |  27 | Philadelphia  |  7000018 | Emma    |  32 | San Diego     |  6600019 | James   |  26 | Raleigh       |  6300020 | Avery   |  35 | Tampa         |  71000
(20 rows)

总结

该方法不仅可以应用到数据库对数据库之间，也可以应到，数据库对文件路径下。在postgresql嵌入python代码 其实可以替换掉一些中间件的使用。可控性，定制性也会更强。

这篇关于POSTGRESQL中ETL、fdw的平行替换的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/603176。

相关文章

Python正则表达式匹配和替换的操作指南

Python正则表达式匹配和替换的操作指南

《Python正则表达式匹配和替换的操作指南》正则表达式是处理文本的强大工具,Python通过re模块提供了完整的正则表达式功能,本文将通过代码示例详细介绍Python中的正则匹配和替换操作,需要的朋... 目录基础语法导入re模块基本元字符常用匹配方法1. re.match() - 从字符串开头匹配2.
阅读更多...
SpringBoot全局域名替换的实现

SpringBoot全局域名替换的实现

《SpringBoot全局域名替换的实现》本文主要介绍了SpringBoot全局域名替换的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录 项目结构⚙️ 配置文件application.yml️ 配置类AppProperties.Ja
阅读更多...
PostgreSQL简介及实战应用

PostgreSQL简介及实战应用

《PostgreSQL简介及实战应用》PostgreSQL是一种功能强大的开源关系型数据库管理系统,以其稳定性、高性能、扩展性和复杂查询能力在众多项目中得到广泛应用,本文将从基础概念讲起,逐步深入到高... 目录前言1. PostgreSQL基础1.1 PostgreSQL简介1.2 基础语法1.3 数据库
阅读更多...
C#高效实现Word文档内容查找与替换的6种方法

C#高效实现Word文档内容查找与替换的6种方法

《C#高效实现Word文档内容查找与替换的6种方法》在日常文档处理工作中,尤其是面对大型Word文档时,手动查找、替换文本往往既耗时又容易出错,本文整理了C#查找与替换Word内容的6种方法,大家可以... 目录环境准备方法一：查找文本并替换为新文本方法二：使用正则表达式查找并替换文本方法三：将文本替换为图
阅读更多...
Python批量替换多个Word文档的多个关键字的方法

Python批量替换多个Word文档的多个关键字的方法

《Python批量替换多个Word文档的多个关键字的方法》有时,我们手头上有多个Excel或者Word文件,但是领导突然要求对某几个术语进行批量的修改,你是不是有要崩溃的感觉,所以本文给大家介绍了Py... 目录工具准备先梳理一下思路神奇代码来啦！代码详解激动人心的测试结语嘿，各位小伙伴们，大家好！有没有想
阅读更多...
Oracle迁移PostgreSQL隐式类型转换配置指南

Oracle迁移PostgreSQL隐式类型转换配置指南

《Oracle迁移PostgreSQL隐式类型转换配置指南》Oracle迁移PostgreSQL时因类型差异易引发错误,需通过显式/隐式类型转换、转换关系管理及冲突处理解决,并配合验证测试确保数据一致... 目录一、问题背景二、解决方案1. 显式类型转换2. 隐式转换配置三、维护操作1. 转换关系管理2.
阅读更多...
linux批量替换文件内容的实现方式

linux批量替换文件内容的实现方式

《linux批量替换文件内容的实现方式》本文总结了Linux中批量替换文件内容的几种方法,包括使用sed替换文件夹内所有文件、单个文件内容及逐行字符串,强调使用反引号和绝对路径,并分享个人经验供参考... 目录一、linux批量替换文件内容 二、替换文件内所有匹配的字符串 三、替换每一行中全部str1为st
阅读更多...
postgresql使用UUID函数的方法

postgresql使用UUID函数的方法

《postgresql使用UUID函数的方法》本文给大家介绍postgresql使用UUID函数的方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录PostgreSQL有两种生成uuid的方法。可以先通过sql查看是否已安装扩展函数，和可以安装的扩展函数
阅读更多...
PostgreSQL中rank()窗口函数实用指南与示例

PostgreSQL中rank()窗口函数实用指南与示例

《PostgreSQL中rank()窗口函数实用指南与示例》在数据分析和数据库管理中,经常需要对数据进行排名操作,PostgreSQL提供了强大的窗口函数rank(),可以方便地对结果集中的行进行排名... 目录一、rank()函数简介二、基础示例：部门内员工薪资排名示例数据排名查询三、高级应用示例1. 每
阅读更多...
PostgreSQL的扩展dict_int应用案例解析

PostgreSQL的扩展dict_int应用案例解析

《PostgreSQL的扩展dict_int应用案例解析》dict_int扩展为PostgreSQL提供了专业的整数文本处理能力,特别适合需要精确处理数字内容的搜索场景,本文给大家介绍PostgreS... 目录PostgreSQL的扩展dict_int一、扩展概述二、核心功能三、安装与启用四、字典配置方法
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2