Сценарии применения Greenplum PXF для интеграции с Data Lake, OLTP, Clickhouse / Habr

Привет! Меня зовут Артемий Козырь, и я Analytics Engineer в Wheely.

Популярность массивно-параллельных СУБД (MPP) для решения аналитических задач неукоснительно растет. Сегодня хотел бы поговорить о широко распространенной СУБД Greenplum и, в частности, о Platform Extension Framework (PXF) - расширении, с помощью которого открываются почти неограниченные возможности интеграции с множеством внешних систем и форматов данных.

В этой публикации Вас ждет:

Основные возможности PXF, конфигурация, способы оптимизации.
Организация Extract - Load с помощью PXF (Data Lake / OLTP).
Объединение локальных и внешних таблиц в запросах (Federated queries).
Запись данных во внешние системы (Clickhouse).

Базовая идея Greenplum PXF

Заключается в том, чтобы задействовать обработку данных на стороне внешних систем, тем самым обеспечивая пользователю возможность обращения ко всей истории данных вне зависимости от места хранения, но без дополнительных трат ресурсов и времени на полную репликацию.

Демонстрационный сценарий описывает источники, отражающие данные по продажам за период в несколько лет. Операционные данные (например, текущий месяц - hot data) хранятся в OLTP MySQL, в Greenplum хранятся данные для аналитической отчетности (например, последний год-два – warm data), а в AWS S3 архивируются данные за более ранние периоды (cold data).

Таким образом, запрос на вывод суммы продаж с группировкой помесячно может быть распределен на 3 системы и выполнен параллельно. При этом пользователь обращается к одной таблице.

Заявлена поддержка доступа к данным следующих систем:

Hadoop / Hive / HBase
AWS S3 / Google Cloud Storage / Azure Blob Storage / MinIO
Реляционные СУБД (через JDBC)
Network file systems

И следующих форматов хранения:

Text (plain, delimited, embedded line feeds)
JSON
Avro, AvroSequenceFile
SequenceFile, RCFile
ORC / Parquet

В рамках этой публикации я буду использовать управляемый сервис от Я.Облака – Yandex Managed Service for Greenplum®. Конфигурация кластера: 2 x Master + 2 x Segment хоста на базе s2.micro (2 vCPU, 100% vCPU rate, 8 GB RAM).

Managed Service for Greenplum уже включает расширение PXF и его базовую конфигурацию. В случае использования Self-hosted Greenplum, все шаги установки и конфигурации придется проделать самостоятельно:

Соблюдение ряда требований перед установкой.
Загрузка PXF Package.
Установка Package на хостах.
Инициализация и запуск сервиса PXF.

Пошаговые действия описаны в официальной документации, и я не буду останавливаться на этом подробно.

PXF - расширение для работы с внешними данными через EXTERNAL TABLEs

PXF реализует протокол для СУБД Greenplum, который можно использовать при создании внешних таблиц.

Синтаксис комадны CREATE EXTERNAL TABLE с протоколом pxf выглядит следующим образом:

CREATE [WRITABLE] EXTERNAL TABLE <table_name>
( <column_name> <data_type> [, ...] | LIKE <other_table> )
LOCATION('pxf://<path-to-data>?PROFILE=<profile_name>[&SERVER=<server_name>][&<custom-option>=<value>[...]]')
FORMAT '[TEXT|CSV|CUSTOM]' (<formatting-properties>)
;

Краткое пояснение к ключевым параметрам конфигурации:

С помощью PXF можно осуществлять как чтение из внешних источников, так и запись данных во внешние системы (ключевое слово WRITABLE). Также возможно объединение данных из разных источников в одном запросе (т.н. federated queries).

Отдельно хочу заострить внимание на поддержке Filter pushdown и Column Projection.

Filter pushdown позволяет применить ограничение на читаемые строки из выражения WHERE запроса SELECT на стороне источника данных, тем самым значительно снижая нагрузку и объем данных, передаваемых по сети. Например, это может быть использовано в обращениях к внешним СУБД, исключению партиций в Hive (partition pruning), чтению групп строк в колоночных форматах (ORC, Parquet).

Column Projection означает то, что только запрошенные колонки SELECT-запроса буду возвращены из внешних систем. Например, при запросе 5 колонок из 100 возможных в файлах формата Parquet в результате запроса вернутся (будут переданы по сети) только 5, что кратное уменьшает объем данных.

Интеграция с Data Lake (S3 / GCS / HDFS / MinIO)

Создадим EXTERNAL TABLE, указывающую на данные в объектном хранилище:

-- 1. Create EXTERNAL TABLE pointing to S3 (Text file)
DROP EXTERNAL TABLE IF EXISTS src_orders ;
CREATE EXTERNAL TABLE src_orders(
O_ORDERKEY BIGINT,
O_CUSTKEY INT,
O_ORDERSTATUS CHAR(1),
O_TOTALPRICE DECIMAL(15,2),
O_ORDERDATE DATE,
O_ORDERPRIORITY CHAR(15),
O_CLERK  CHAR(15),
O_SHIPPRIORITY INTEGER,
O_COMMENT VARCHAR(79)
)
LOCATION ('pxf://otus-dwh/tpch-dbgen/orders.csv?PROFILE=s3:csv&accesskey=<>&secretkey=<>&endpoint=storage.yandexcloud.net'
)
FORMAT 'TEXT'
(DELIMITER '|')
;

Явно подчеркну важность корректной конфигурации для чтения файла, а именно:

Адрес в бакете – otus-dwh/tpch-dbgen/orders.csv
Профиль для чтения формата файла – PROFILE=s3:csv
Наличие ключей для доступа к бакету – accesskey=<>&secretkey=<>
Указание endpoint для S3-подобных хранилищ, например endpoint=storage.yandexcloud.net
Параметры для чтения конкретных форматов – для текста это (DELIMITER '|')

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

Всего в файле содержится ровно 15М строк.

-- 2. Count rows
SELECT count(1) FROM src_orders ; -- 15000000 ROWS

Теперь пробуем запустить аналитический запрос и посмотреть план его выполнения:

-- 3. Run OLAP query
EXPLAIN ANALYZE
SELECT
    DATE_TRUNC('month', O_ORDERDATE) AS order_year
    ,count(1) AS num_orders
FROM src_orders
WHERE O_ORDERSTATUS = 'P'
GROUP BY 1
ORDER BY 1 ASC 
;

На выполнение запроса потребовалось 21.5 секунды, львиная доля времени была затрачена на чтение данных в S3. Даже при наличии фильтра WHERE потребовалось полное чтение файла, так как текстовые файлы не поддерживают predicate pushdown.

Далее с помощью WRITEABLE таблицы запишем эти же данные обратно в S3, но уже в колоночном формате Parquet:

-- 4. Write data back to S3 in columnar format (Parquet)
DROP EXTERNAL TABLE IF EXISTS trg_orders ;
CREATE WRITABLE EXTERNAL TABLE trg_orders(
    O_ORDERKEY BIGINT,
    O_CUSTKEY INT,
    O_ORDERSTATUS CHAR(1),
    O_TOTALPRICE DECIMAL(15,2),
    O_ORDERDATE DATE,
    O_ORDERPRIORITY CHAR(15), 
    O_CLERK  CHAR(15), 
    O_SHIPPRIORITY INTEGER,
    O_COMMENT VARCHAR(79)
)
LOCATION ('pxf://otus-dwh/tpch-dbgen-parquet/orders?PROFILE=s3:parquet&accesskey=<>&secretkey=<>&endpoint=storage.yandexcloud.net'    
)
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');
INSERT INTO trg_orders SELECT * FROM src_orders ORDER BY O_ORDERSTATUS, O_ORDERDATE
;

Запустим тот же самый OLAP-запрос к новой таблице и сравним результаты:

В этот раз запрос выполнился за 4 секунды (в 5 раз быстрее), с учетом predicate pushdown (WHERE O_ORDERSTATUS = 'P') и column projection (фактически читали 1 колонку - O_ORDERDATE), поддерживаемых форматом Parquet.

Интеграция с OLTP СУБД (PostgreSQL)

В этом сценарии перейдем к работе с внешними OLTP-базами, которые динамично наполняются новыми данными и меняют уже имеющиеся. Предположим, что в нашем случае это статистические данные о рекламных кампаниях:

-- 1. Create EXTERNAL TABLE pointing to PostgreSQL table
DROP EXTERNAL TABLE IF EXISTS src_direct_ads_facts ;
CREATE EXTERNAL TABLE src_direct_ads_facts (
    id serial ,
    account_id int4 ,
    dates_id int4 ,
    sites_id int4 ,
    traffic_id int4 ,
    device varchar(16) ,
    impressions_context int4 ,
    impressions_search int4 ,
    impressions int4 ,
    clicks_context int4 ,
    clicks_search int4 ,
    clicks int4 ,
    cost_context numeric(18, 2) ,
    cost_search numeric(18, 2) ,
    "cost" numeric(18, 2) ,
    average_position numeric(18, 2) ,
    average_position_clicks numeric(18, 2) ,
    campaigns_id int4 ,
    ads_id int4 ,
    adgroups_id int4 ,
    bounces int4
)
LOCATION ('pxf://public.direct_ads_facts?PROFILE=JDBC&JDBC_DRIVER=org.postgresql.Driver&DB_URL=jdbc:postgresql://<hostname>:6432/<database>&USER=<username>&PASS=<password>')
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_import')
;

В случае JDBC видим уже несколько другой набор параметров:

Схема и имя таблицы – public.direct_ads_facts
Профиль для чтения – PROFILE=JDBC
Класс JDBC-драйвера – JDBC_DRIVER=org.postgresql.Driver
DB URI – DB_URL=jdbc:postgresql://<hostname>:6432/<database>&USER=<username>&PASS=<password>'

Будем считать, что необходимо регулярно получать данные из системы-источника в Greenplum. Сделать это можно двумя способами:

Копирование всей таблицы при каждом запросе.
Организация инкрементальной загрузки.

Первый способ обладает очевидными недостатками – чрезмерная нагрузка на источник и копирование одних и тех же данных каждый раз. Инкрементальный способ загрузки представляется гораздо более эффективным, но для него понадобятся способы выделения дельты (изменений), например, монотонно возрастающий ключ:

-- 1. Initialize
CREATE TABLE direct_ads_facts AS SELECT * FROM src_direct_ads_facts ;
-- 2. Incremental load
INSERT INTO direct_ads_facts SELECT * FROM src_direct_ads_facts
WHERE id > (SELECT MAX(id) FROM direct_ads_facts) ;

Использование федеративных запросов (Federated Queries)

Теперь представьте ситуацию, когда огромную таблицу фактов нужно склеить с маленьким, но часто меняющимся справочником. Реальный пример - это справочник названий рекламных кампаний, которые всегда имеют одинаковый идентификатор, но периодически меняют свои названия (label).

Для этого идеально подойдет PXF с возможностью обращения к источнику src_direct_campaigns для получения самых актуальных значений из справочника:

-- 2.7. Create Data Mart
CREATE TABLE direct_search_facts AS
SELECT    
      cf.account_id AS account_id
    , MD5(CONCAT(CAST(cf.id AS varchar), 'yandex.search')) AS id
    , ga.caption AS caption
    , gd.simple_date AS dt
    , cf.device AS device
    , 'yandex.search' AS SOURCE
    , gt.medium AS medium
    , CAST(cp.campaign_id AS Int) AS campaign_id
    , cp.name AS campaign -- Always correct and up-to-date
    , gt.campaign AS traffic_campaign
    , gt.content AS CONTENT
    , gt.keyword AS keyword
    , gs.domain AS DOMAIN
    , gt.landing_page AS landing_page
    , cf.impressions_search AS impressions
    , cf.clicks_search AS clicks
    , cf.cost_search AS COST
FROM direct_ads_facts AS cf
    LEFT JOIN general_accounts AS ga
            ON ga.account_id = cf.account_id
    LEFT JOIN general_dates AS gd
            ON gd.id = cf.dates_id
    LEFT JOIN src_direct_campaigns AS cp -- ! From PostgreSQL directly
            ON cp.id = cf.campaigns_id
    LEFT JOIN general_sites AS gs
            ON gs.id = cf.sites_id
    LEFT JOIN general_traffic AS gt
            ON gt.id = cf.traffic_id
;

Запись данных в Clickhouse для сверхбыстрого чтения

Полученная широкая таблица-витрина содержит все данные, чтобы задавать многочисленные аналитические вопросы. Однако, порой даже Greenplum может не справляться с множеством сложных аналитических запросов, существенно теряя в скорости и интерактивности. На помощь приходит сверхбыстрый Clickhouse! Наша задача - переместить готовую витрину данных туда, где с ней можно будет работать в режиме пинг-понг, вне зависимости от объема данных.

Для начала на стороне приемника необходимо создать пустую таблицу (схему):

CREATE TABLE direct_search(
	  account_id Int8
    , id String
	, caption String
	, dt String
	, device String
	, "source" String
	, medium String
	, campaign_id Int8
	, campaign String
	, traffic_campaign String
	, "content" String
	, keyword String
	, "domain" String
	, landing_page String
	, impressions Int8
	, clicks Int8
	, "cost" Float32
)
ENGINE = MergeTree()  
ORDER BY (dt)
;

После этого мы готовы к записи данных в Clickhouse с помощью PXF:

DROP EXTERNAL TABLE IF EXISTS trg_ch_direct_search ;

CREATE WRITABLE EXTERNAL TABLE trg_ch_direct_search(
account_id int4 ,
id varchar(128) ,
caption varchar(128) ,
dt varchar(128) ,
device varchar(16) ,
"source" varchar(1024) ,
medium varchar(1024) ,
campaign_id int4 ,
campaign varchar(1024) ,
traffic_campaign varchar(1024) ,
"content" varchar(1024) ,
keyword varchar(1024) ,
"domain" varchar(1024) ,
landing_page varchar(1024) ,
impressions int4 ,
clicks int4 ,
"cost" numeric(18, 2)
)
LOCATION ('pxf://direct_search?PROFILE=JDBC&JDBC_DRIVER=ru.yandex.clickhouse.ClickHouseDriver&DB_URL=jdbc:clickhouse://<hostname>:8123/default')
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export')
;

INSERT INTO trg_ch_direct_search
SELECT
account_id
,id
,caption
,dt
,device
,"source"
,medium
,campaign_id
,campaign
,traffic_campaign
,"content"
,keyword
,"domain"
,landing_page
,impressions
,clicks
,"cost"
FROM direct_search
;

Ничего сложного, несколько простых действий, немного терпения и готово. Теперь с этой витриной можно работать в Clickhouse, который, как известно, не тормозит!

Проблемы и трудности

В рамках исследования и подготовки материала я столкнулся с несколькими трудностями, на которые хотел бы обратить ваше внимание:

В случае работы с S3 необходимо указывать ключи доступа accesskey=<>&secretkey=<> даже для публично доступных buckets.
Для работы PXF в Yandex Managed Service for Greenplum необходимо включить Egress NAT для подсети хостов.

С недавнего времени на сайте с документацией появилось релевантное сообщение:

Для подключения к внешним источникам необходимо включить NAT в интернет для подсети, в которой расположен кластер Managed Service for Greenplum®.

В противном случае вы будете получать ошибку с таймаутом SQL Error [08000]: ERROR: PXF server error : Could not obtain datasource for server default and PoolDescriptor

Отсутствие SSL-соединения для JDBC.

«PXF» не поддерживает «SSL-соединение» на уровне параметров драйвера ClickHouse-JDBC.

Workaround: Вы можете оставить в кластере ClickHouse один хост без публичного доступа и к нему обращаться из Greenplum.

Умение строить комплексные решения, отвечающие на запросы бизнеса

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

Именно эти аспекты я держал в уме, когда работал над программами курсов Analytics Engineer, Data Engineer, DataOps Engineer в OTUS.

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

В комментарии поделитесь, с каким кейсом использования PXF удалось поработать, либо как планируете применять?