43 lines
1.0 KiB
PL/PgSQL
43 lines
1.0 KiB
PL/PgSQL
create schema stg;
|
|
|
|
CREATE TABLE stg.samplekafka2postgres (
|
|
dttm timestamptz NULL,
|
|
txt text NULL
|
|
);
|
|
|
|
|
|
create schema ods;
|
|
|
|
CREATE TABLE ods.samplekafka2postgres (
|
|
id serial4 NOT NULL,
|
|
ins_dttm timestamptz default now() NOT NULL,
|
|
dttm timestamptz NULL,
|
|
txt text NULL,
|
|
CONSTRAINT samplekafka2postgres_pkey PRIMARY KEY (id)
|
|
);
|
|
|
|
create or replace procedure ods.load_samplekafka2postgres()
|
|
language plpgsql
|
|
as $$
|
|
begin
|
|
-- перенос содержимого таблицы в ods полностью
|
|
insert into ods.samplekafka2postgres(dttm, txt)
|
|
select dttm, txt
|
|
from stg.samplekafka2postgres;
|
|
|
|
-- удаление делаем в процессе NiFi
|
|
-- delete from stg.samplekafka2postgres;
|
|
|
|
commit;
|
|
end; $$;
|
|
|
|
-- Примеры запросов
|
|
select * from ods.samplekafka2postgres;
|
|
|
|
delete from stg.samplekafka2postgres;
|
|
delete from ods.samplekafka2postgres where ins_dttm < now() - interval '10 minute';
|
|
|
|
select * from stg.samplekafka2postgres;
|
|
select * from ods.samplekafka2postgres order by id desc;
|
|
select count(*) from ods.samplekafka2postgres;
|