SanSYS Dec 10 2018 at 06:34

PostgreSQL: PipelineDB — агрегирующие запросы в режиме реального времени

5 min

11K

PostgreSQL*

Tutorial

+25

Comments 24

UFO just landed and posted this here

SanSYS Dec 10 2018 at 07:51

Кажется расширения под 9.6 нет, но я не сильно искал

PipelineDB currently supports PostgreSQL versions 10.1, 10.2, 10.3, 10.4, 10.5, and 11.0

Как быть — заюзать уже пг 10 )
Даже 11-й зарелизился
Ну или если решили, что нужно использовать PipelineDB, а хотите продолжать сидеть на 9.6, то всегда можно поднять ещё один свежий Postgres и там развлекаться

SanSYS Dec 10 2018 at 07:51

del

Tantrido Dec 10 2018 at 08:22

Интересно: а это расширение может помочь, если данные будут в основном в JSONB поле?

SanSYS Dec 10 2018 at 08:59

Типа того?

select minute(eventtime) as time, 
       sum((data->>'value')::integer) as sum
from table1

Не вижу проблем с таким вариантом, хотя не пробовал

Тут же суть в том, что расширение организует поточную обработку данных
т.е.: приходят данные, возможно накапливаются в каком-то буфере, в фоне выполняется запрос из вьюхи над новыми данными, далее — результат вставляется в результирующую таблицу, либо — дополяет старые данные

И чисто теоретически не сильно важно, какого типа данные гоняются от потока до вьюхи, потому что вы сами же с ними и работаете весьма прозрачным образом

Например если у вас в jsonb хранится массив объектов и хотите именно его обрабатывать «в потоке», то можно сделать трансформацию разложив массив на строки и… работать как с обычной таблицей

Tantrido Dec 10 2018 at 09:03

А можно JSONB разложить в представление (view)?

SanSYS Dec 10 2018 at 10:19

Определённо да, вечером скину пример

Envek Dec 10 2018 at 17:10

Один JSON-объект можно разложить в запись с помощью jsonb_to_record, а массив — с помощью jsonb_to_recordset. См. официальную документацию и мой пример.

SanSYS Dec 10 2018 at 18:03

Как подсказали выше, вот так

select * 
from jsonb_to_recordset(
        '[{"a":1,"b":"foo"},{"a":"2","c":"bar", "v":"2018-04-01"}]'::jsonb
) as x(a int, b text, v date, not_exists time);

Tantrido Dec 11 2018 at 00:48

Это простенький JSON, а если сложный иерархический, но структура более менее постоянна.

Envek Dec 12 2018 at 19:12

Можно вытаскивать нужный кусок JSONB и скармливать его это же функции, на ходу подправляя, если требуется.
Вот, например, если у нас есть таблица listings, у которой есть jsonb-поле raw, где-то в недрах которого лежит нужный нам массив (а иногда не массив):

SELECT id, v."SKU"
FROM 
  listings, 
  jsonb_to_recordset(
    CASE jsonb_typeof(raw->'Variations'->'Variation')
    WHEN 'array' THEN raw->'Variations'->'Variation'
    ELSE jsonb_build_array(raw->'Variations'->'Variation')
    END
  ) AS v("SKU" varchar)
WHERE raw IS NOT NULL 
  AND raw->'Variations'->'Variation' IS NOT NULL

UncleAndy Dec 10 2018 at 11:10

Т.е. эта система потокобезопасная и лишена дедлоков при наличии вставок через несколько подключений к БД?

SanSYS Dec 10 2018 at 18:21

Не знаю
Уверен, что при простом использовании стримов и вьюх — всё будет ок
Но если использовать вызовы различных кастомных методов, то внутри можно написать любуую логику и да — монжо словить классический дедлок, всё как всегда
У них в репе issues имеют метку deadlock, но они скорее не к экстеншену для postgres

SergeyGershkovich Dec 10 2018 at 11:58

А где в статье связь между запросом

SELECT hour(datetime), somename, count(*), sum(somemetric)
from table
where datetime > :monthAgo
group by 1, 2
order by 1 desc, 2

и стримом


CREATE FOREIGN TABLE flow_stream (
    dtmsk timestamp without time zone,
    action text,
    duration smallint
) SERVER pipelinedb;

Я так понимаю, в «table» нужно триггер или правило добавлять, чтобы заливать данные в «flow_stream»?

VJean Dec 10 2018 at 12:17

PipelineDB можно рассматривать как таблицы с интегрированными триггерами.

SergeyGershkovich Dec 10 2018 at 13:00

Мне очень понравился первый абзац:

Вас когда-либо просили посчитать количество чего-то на основании данных в бд за последний месяц, сгруппировав результат по каким-то...

Что если у меня уже есть Postgres DB со своими таблицами, к которым требуется писать множество агрегатных запросов. Я так понимаю, чтобы использовать PipelineDB, я должен переместить таблицы в PipelineDB или написать некие правила для отправки данных из своих таблиц в PipelineDB? — этих правил в статье я не увидел.

VJean Dec 10 2018 at 15:48

Достаточно детальный пример с примечаниями: Streaming Databases & PipelineDB: Likes Use Case

В статье много чего не описано по сравнению с официальной документацией, в том числе и про докер, т.к. готовый образ уже собран.

SergeyGershkovich Dec 10 2018 at 19:37

Почитал документацию, для себя определил PipelineDB, как механизм агрегирующих счетчиков в режиме реального времени. Счетчик от потока отключил, агрегированные данные потерял.
Идея интересная, но не для случая уже хранимых данных. Удобно для данных, которые только планируется сохранять, с учетом, что счетчики должны быть включены заранее.

UFO just landed and posted this here

Tatikoma Dec 10 2018 at 14:19

curl -s http://download.pipelinedb.com/apt.sh | bash

Очень сильно удивился увидев такую конструкцию. Открыл документацию pipelinedb и увидел там ровно такую же строчку. Выглядит очень грустно.

Pinkerton42 Dec 10 2018 at 16:27

Угу. В нем безапеляционно мелькают строки «apt-get install -y ...» без всяких вопросов. За такое обычно по рукам бьют.

Tatikoma Dec 10 2018 at 17:20

Это проблемы разного уровня, если я ничего не путаю.
apt-get install -y в худшем случае сломает вам систему, если у вас нет косячных репозиториев.
curl -s http | bash — в худшем случае установит вирус, который спросит права на рута и вы ему их сами предоставите

Pinkerton42 Dec 10 2018 at 18:45

Неважно какого уровня: это — проблема. И считаю, что да, за такое надо бить по рукам.

neenik Dec 10 2018 at 14:58

Поздно postgresql, поздно. OLAP+SQL=ClickHouse.
У нас у самих до сир пор осталось пяток самописных агрегаций в постгресе. Характеристика одним словом — неудобно.

Show the best of all time