PG存储函数一则（1）——拆分json字段

PG的json字段是很好用的，但是，领导说是弄数据仓库还是什么的，非要把json字段拆分成15张表，脑残的需求昂，但是还是要做啊。贴存储过程的示例主要是展示存储过程的写法，逻辑性还是比较强的，存储函数，啊存储函数存储函数，学MySQL比较容易称呼成存储过程，PG里面是存储函数哈，存储函数不像java等开发语言那样有很多可以调用的函数，很多情况下要手动调整。大家借鉴啦。

原始表：org_raw_gs,org_raw_gz,org_raw_heb,org_raw_hen这样的表，org_raw_省份的后缀，表结构全部是一样的：



我们看一条记录：



主要是body字段，body是一个jsonb字段：

{“基本信息”: {“”: “”, “顺序”: [“企业名称”, “营业执照注册号”, “注册地址”, “工商登记机关”, “归类行业”, “注册资本（万元）”, “市场主体类型”, “经营范围”, “一般经营范围”, “成立日期”, “营业期限起始日期”, “营业期限到期日期”, “注册地址联系电话”, “法定代表人（负责人）姓名”, “企业类型”, “注册资金币种”, “核准日期”, “登记状态”, “经营状态”, “”], “企业名称”: “泰安路迎丰粮油店”, “企业类型”: “个体工商户”, “归类行业”: “”, “成立日期”: “2009-03-09 00:00:00”, “核准日期”: “2014-01-07 00:00:00”, “注册地址”: “金川区泰安路”, “登记状态”: “吊销”, “经营状态”: “”, “经营范围”: “米、面粉、植物油零售。”, “一般经营范围”: “”, “工商登记机关”: “金昌市工商行政管理局金川分局”, “市场主体类型”: “”, “注册资金币种”: “万人民币”, “营业执照注册号”: “620300600029787”, “注册地址联系电话”: “”, “注册资本（万元）”: “.5”, “营业期限到期日期”: “2013-03-08 00:00:00”, “营业期限起始日期”: “2009-03-09 00:00:00”, “法定代表人（负责人）姓名”: “王治秀”}}

把这个长字段贴到json.cn中可以看出他是一个两层的json结构，其中有的值是数组，现在我们就是做循环，把键值对拆成字段和值：

---------------------
--创建存储过程---------
---------------------
-- Function: public.split_json(character varying)

-- DROP FUNCTION public.split_json(character varying);

CREATE OR REPLACE FUNCTION split_json(tb_name varchar) RETURNS int AS
$$

DECLARE
xiangqing RECORD;
keyvalue_1 RECORD;
ifcolumn_exist int;
to_tb_name varchar(50);
text_var1 text;
text_var2 text;
text_var3 text;
BEGIN
to_tb_name:=concat('to_',tb_name);
EXECUTE 'drop table if exists '||to_tb_name;
EXECUTE 'create table '||to_tb_name||'(id varchar(100),org_id varchar(50),name varchar(80),time varchar(80),primary key(id))';
<<loop_to_table>>
FOR xiangqing IN EXECUTE 'SELECT id,org_id,name,body,time FROM '||tb_name::regclass||' where body is not null' LOOP
IF xiangqing.body ::varchar = '' THEN
RAISE NOTICE 'body is 空串';
ELSE
EXECUTE 'insert into '
||to_tb_name::regclass
||'(id,org_id,name,time) values ($1,$2,$3,$4)'
USING xiangqing.id,xiangqing.org_id,xiangqing.name,xiangqing.time;
<<loop_outer_key>>
FOR keyvalue_1 IN (select key,value::varchar from jsonb_each(xiangqing.body::jsonb)) LOOP
RAISE NOTICE '外层键--> %',keyvalue_1.key::varchar;
IF trim(keyvalue_1.key::varchar)<>'' THEN
--查看字段存在否，如果不存在，alter table添加字段
select count(*) INTO ifcolumn_exist from information_schema.columns where table_name = to_tb_name and column_name=keyvalue_1.key::varchar;
IF (ifcolumn_exist=0) THEN
--字段不存在就修改表结构
EXECUTE 'alter table '
|| to_tb_name::regclass
||' add column "'
|| keyvalue_1.key
|| '" text';
END IF;
--然后更新字段值
EXECUTE 'update '
|| to_tb_name::regclass
||' set "'
|| keyvalue_1.key
|| '"=$1 where id=$2'
USING keyvalue_1.value,xiangqing.id;

--尝试去分隔var_value，有异常则抛出，能分隔则继续往下处理
--创建一个子块
DECLARE keyvalue_2 RECORD;
BEGIN
<<sub_block>>
FOR keyvalue_2 IN (select key,value from jsonb_each(keyvalue_1.value::jsonb)) LOOP
RAISE NOTICE '内层键--> %',keyvalue_2.key::varchar;
IF trim(keyvalue_2.key::varchar) <> '' THEN
--查看字段存在否，如果不存在，alter table添加字段
select count(*) INTO ifcolumn_exist from information_schema.columns where table_name = to_tb_name and column_name=keyvalue_2.key::varchar;
IF (ifcolumn_exist=0) THEN
--字段不存在就修改表结构
EXECUTE 'alter table '
|| to_tb_name::regclass
||' add column "'
|| keyvalue_2.key
|| '" varchar(200)';
END IF;
EXECUTE 'update '
|| to_tb_name::regclass
||' set "'
|| keyvalue_2.key
|| '"=$1 where id = $2'
USING trim('"' from keyvalue_2.value::varchar),xiangqing.id;

END IF;--结束判断keyvalue_2  empty

END LOOP sub_block;
EXCEPTION
WHEN  OTHERS  THEN
GET STACKED DIAGNOSTICS text_var1 = MESSAGE_TEXT,
text_var2 = PG_EXCEPTION_DETAIL,
text_var3 = PG_EXCEPTION_HINT;
RAISE NOTICE '% accurs exception\n %\n%\n%', keyvalue_1.key::varchar,
text_var1,text_var2,text_var3;
continue;
END;
END IF;--结束判断键值是否empty
--退出条件
END LOOP loop_outer_key;
END IF; --结束判断body是否为空串
--退出条件
END LOOP loop_to_table;

return 1;
END;

$$ LANGUAGE plpgsql ;

--调用存储过程
select split_json('org_raw_gs');
--有一些non json object的报错信息忽略就好

结果得到很多很多的字段：



然后根据需要的信息拆分成15张表，有这样的一个需求就是有一个数组字段就是表的主体内容，比如说”工商公示信息-变更信息”就是一个数组字段，我们看一下：

[[“变更事项”,”变更前内容”,”变更后内容”,”变更日期”],[“负责人变更”,”付伯权”,”范石涛”,”2015年8月28日”],[“经营范围变更”,”办理人民币存款 贷款、结算业务；办理票据贴现业务；代理发行、兑付政府债券；代理发行金融债券；代理收付款项；外汇存款；外汇贷款；外汇汇款；外币兑换，结汇、售汇；国际结算；外汇票据的承兑和贴现；资信调查、咨询、见证业务；通过上级行办理代客外汇买卖；外汇信用卡的发放，代理国外信用卡付款；个人黄金买卖业务；经中国银行业监督管理委员会等监管部门批准的其他业务;安装工程保险、机动车辆保险、家庭财产保险、建筑工程保险、健康保险、企业财产保险、人寿保险、意外伤害保险（保险兼业代理许可证有效期至2013年12月16日） “,”办理人民币存款 贷款、结算业务；办理票据贴现业务；代理发行、兑付政府债券；代理发行金融债券；代理收付款项；代理保险业务（保险法律法规和行政规章制度许可范围内的险种，保险兼业代理许可证有效期至2016年12月16日）；外汇存款；外汇贷款；外汇汇款；外币兑换，结汇、售汇；国际结算；外汇票据的承兑和贴现；资信调查、咨询、见证业务；通过上级行办理代客外汇买卖；外汇信用卡的发放，代理国外信用卡付款；个人黄金买卖业务；经中国银行业监督管理委员会等监管部门批准的其他业务。（依法须经批准的项目，经相关部门批准后方可开展经营活动）”,”2014年9月19日”]]

我们看到这个数组第一行是键，从第二行开始是对应的值，目标表结构如下：

create table gs_penalty_info (
auto_id serial,
id varchar(80)  ,   -- '唯一id' ,
corp_id varchar(30) ,   -- '注册号' ,
corp_name varchar(100) ,   -- '名称' ,
--no varchar(6) null,   -- '序号' ,
--case_no varchar(30) null,   -- '行政处罚决定书文号' ,
--case_type varchar(32) null,   -- '违法行为类型' ,
--case_content varchar(1024) null,   -- '行政处罚内容' ,
--case_org varchar(100) null,   -- '作出行政处罚决定机关名称' ,
--case_date timestamp null,   -- '作出行政处罚决定日期' ,
--case_detail varchar(20148) null,   -- '详情' ,
--create_date timestamp ,   -- '创建时间' ,
--create_by varchar(32) ,   -- '创建者' ,
--update_date timestamp ,   -- '更新时间' ,
--update_by varchar(32) ,   -- '更新者' ,
--source varchar(32) ,   -- '数据来源' ,
primary key (auto_id)
);

其中一些字段我们先不创建，在拆分的时候添加字段，比较通用的做法是这样：

CREATE OR REPLACE FUNCTION split_array(tmp_tb_name varchar,target_tb varchar,colu_name varchar,begin_xh int, end_xh int) RETURNS int AS
$$

--------过程参数说明：
---------tmp_tb_name：中间表名称，比如 to_org_raw_heb
---------target_tb:目标表，15张表的某一张，
---------colu_name:数组字段的名称，比如："动产抵押登记信息"这个字段
---------begin_xh，end_xh:在中间表中加了一个auto_incr的自增字段，可以小区间小区间的调用，防止数据量过大更新不成功

--------函数主体
DECLARE
target RECORD;
one_dimen int;
two_dimen int;
i int;
j int;
split_zd varchar;
return_i int default 0;
ifcol_exist int default 0;
iftb_exist int default 0;
seq_name varchar default '';
province varchar(5);
text_var1 text;
text_var2 text;
text_var3 text;
BEGIN
province:=substring(tmp_tb_name from 12 for 4);
--如果表不存在就创建，已存在就算了;
EXECUTE 'select count(*) from pg_class where relname='
||quote_literal(target_tb)
INTO iftb_exist;
IF iftb_exist=0 THEN
EXECUTE 'create table '||target_tb||'(auto_id serial, id varchar(80), corp_id varchar(30),corp_name varchar(100), province varchar(5),primary key (auto_id))';
END IF;
<<record_loop>>
FOR target IN EXECUTE 'select id,org_id,name,'||quote_ident(colu_name) || ' as gd_info,auto_incr  from '
|| tmp_tb_name::regclass
|| ' where  auto_incr BETWEEN '
|| begin_xh ||' and '||end_xh ||' and org_id is not null and org_id<>'
|| quote_literal('')||' and '
||quote_ident(colu_name)
||' is not null'
LOOP
RAISE NOTICE '第%行:',target.auto_incr::varchar;
split_zd=replace(replace(target.gd_info,'[','{'),']','}');
one_dimen=array_length((replace(replace(split_zd,'[','{'),']','}'))::text[][],1);
two_dimen=array_length((replace(replace(split_zd,'[','{'),']','}'))::text[][],2);
<<addco>> FOR j IN 1..two_dimen LOOP
EXECUTE 'select count(*)  from information_schema.columns where table_name = $1 and column_name = $2'
INTO ifcol_exist
USING target_tb,(split_zd::text[][])[1][j];
IF (ifcol_exist=0) THEN
--字段不存在就修改表结构
EXECUTE 'alter table '
||target_tb::regclass
||' add column "'
|| (split_zd::text[][])[1][j]
|| '" text';
END IF;
EXIT addco WHEN j=two_dimen;
END LOOP addco;
seq_name=concat(target_tb,'_auto_id_seq');
<<data_in>>  For i in 2..one_dimen LOOP
EXECUTE 'insert into '
||target_tb::regclass
||'(ID,CORP_ID,CORP_NAME,province,"'
|| (split_zd::text[][])[1][1]
|| '") values ($1,$2,$3,$4,$5) '
USING target.id,target.org_id,target.name,province,(split_zd::text[][])[i][1];
<<data_up>>FOR j in 2..two_dimen LOOP
EXECUTE 'update  '
||target_tb::regclass
||' set "'
|| (split_zd::text[][])[1][j]
|| '"= ($1) where auto_id=$2 '
USING (split_zd::text[][])[i][j],currval(seq_name::regclass);
END LOOP data_up;
END LOOP data_in;
return_i = currval(seq_name::regclass);
END LOOP record_loop ;
return return_i;
EXCEPTION WHEN OTHERS THEN
GET STACKED DIAGNOSTICS text_var1 = MESSAGE_TEXT,
text_var2 = PG_EXCEPTION_DETAIL,
text_var3 = PG_EXCEPTION_HINT;
RAISE NOTICE '%\n%\n%',text_var1,text_var2,text_var3;
END;
---------函数主体

$$
LANGUAGE plpgsql;

调用示例：

select split_array('to_org_raw_heb','gs_penalty_info','工商公示信息-行政处罚信息',1,10000);