POSTGRESQL中ETL、fdw的平行替换

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

POSTGRESQL中ETL、fdw的平行替换

01、简介

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

在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)