Привет, Хабр! Мы продолжаем серию статей о проведённой миграции аналитического хранилища данных с платформы Teradata на GreenPlum. В предыдущей статье мы рассказали о нашем опыте и результатах автоматизированного переписывания SQL-скриптов из диалекта Teradata в диалект GreenPlum с помощью реализованного сервиса миграции кода. В этой статье мы расскажем вам о полученном нами опыте и результатах переноса архива данных объёмом более 400 Тб из Teradata в GreenPlum, а также о трудностях и решениях, связанных с этим процессом.

Введение

Перенос архива данных из Teradata в GreenPlum являлся одним из самых критичных и ресурсоёмких этапов проведенной миграции. Хранилище данных содержало исторические данные, которые использовались для аналитики и отчетности. Перенос архива данных должен был быть выполнен с минимальным влиянием на текущую работу хранилища, а также с обеспечением целостности данных.

Как и предполагалось, этот процесс имел свои сложности и ограничения.

Трудности переноса

• Объёмы данных. Наш архив данных в Teradata содержал более 400 Тб данных. Перенос такого количества требовал много ресурсов, времени и памяти.

• Во время переноса все еще выполнялись регулярные загрузки данных в Teradata.

• Отсутствие доступных нам в промышленной среде сервисов гетерогенного доступа между Teradata и Greenplum, обеспечивающих приемлемую скорость передачи.

Мы не могли использовать прямой или косвенный доступ к данным в Teradata из GreenPlum, так как это было бы слишком медленно или нестабильно. Поэтому разработали собственное решение для миграции данных, основанное на использовании промежуточного хранилища в Hadoop, которое позволило нам обойти эти ограничения и использовать все доступные нам инструменты максимально эффективно, достигнув высокой скорости и надёжности миграции данных.

Архитектура решения задачи по переносу данных

Наша архитектура миграции данных хранилища из Teradata в GreenPlum состояла из следующих основных компонентов:

• Исходное хранилище данных, из которого мы переносили данные. Кластер Teradata содержал более 400 Тб данных, распределённых по тысячам таблиц. В кластере был доступен сервис Teradata QueryGrid с коннектором Teradata‑Hive, который позволял нам выгружать данные из Teradata в Hadoop.

• Промежуточное хранилище данных, которое мы использовали для временного хранения переносимых из Teradata в GreenPlum данных. Для загрузки данных из Hadoop в GreenPlum мы использовали фреймворк PXF, который позволял нам получать доступ к данным на Hadoop из GreenPlum.

• Целевое хранилище данных, в которое мы переносили данные GreenPlum.

Схематически наша архитектура решения может быть представлена следующим образом:

Мы уже проводили запись данных из Teradata через QueryGrid на кластер Hadoop через коннектор Teradata-Hive, но практического опыта работы с PXF для записи данных из Hadoop в Greenplum у нас не было. Для проверки производительности фреймворка PXF в типовых задачах для нашего хранилища, пакетной загрузки данных из реплик различных систем источников, хранящихся в Hadoop, решили провести нагрузочное тестирование.

Нагрузочное тестирование PXF по записи данных из Hadoop в Greenplum

Для теста использовали 15 крупных таблиц + 30 средних и малых, запись начали с ограничения в 1 миллион строк, далее кратно увеличивали с 1 до 10 и затем до 100 с использованием запросов вида: SELECT * FROM <TABLE__NAME> LIMIT 100000000;.

На стороне приёмника кластера Greenplum создали external table для обращения к таблицам источника и таблицы-приёмники данных с распределением по случайному ключу без сжатия с аналогичной структурой.

Мы получили следующие результаты:

• Последовательная вставка данных заняла 1340 секунд.

• Параллельная вставка заняла 358 секунд.

• Объём данных на стороне Greenplum после загрузки составил ~325 Гб.

Затем мы проверили скорость передачи данных при увеличении количества параллельных сессий. Мы выполняли параллельные вставки из больших таблиц в 1, 10, 20 и 30 сессий:

Ниже для наглядности прикладываем графики.

Подготовка к переносу

Наше решение для переноса данных основывалось на четырёх видах простых SQL-скриптов.

• Первый набор скриптов создавал таблицы в Hive для хранения данных на Hadoop.

• Второй набор SQL‑скриптов выполнялся на стороне Teradata и вставлял данные из Teradata в Hive‑таблицы на стороне кластера Hadoop через коннектор Teradata‑Hive.

• Третий набор запускался на Greenplum и создавал внешние таблицы для доступа к данным на Hadoop через PXF.

• Четвёртый набор выполнял операцию INSERT в таблицах в Greenplum.

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

Для работы функции потребовалось выполнение следующих условий:

1. Таблицы в Teradata и в GreenPlum имеют одинаковые атрибуты.

2. В кластере Hadoop есть схема в Hive, к которой есть доступ на запись.

3. На стороне Teradata настроен QueryGrid foreign server, позволяющий записывать данные в Hive‑таблицы из п.2.

4. На стороне GreenPlum настроен PXF‑сервер, позволяющий читать данные из parquet-файлов в соответствии с метаданными Hive‑таблиц из Hive metastore..

Функция генерации скриптов в качестве параметром принимала следующие значения:

Результатом выполнения функции были 4 скрипта, включая два запроса INSERT‑SELECT, и DDL-скрипты для создания таблицы в Hive и external table в Greenplum, как было упомянуто выше. Например:

1. Создание таблицы в Hive:

   DROP TABLE IF EXISTS custom_t_propagation_010_src.n010010002kul2_lmt_metric_hist;
   CREATE EXTERNAL TABLE custom_t_propagation_010_src.n010010002kul2_lmt_metric_hist
   (
   crncy_id BIGINT,
   limit_id BIGINT,
   limit_metric_amt DECIMAL(38,16),
   limit_metric_end_dt VARCHAR(10),
   limit_metric_start_dt VARCHAR(32),
   limit_metric_type_id BIGINT,
   sod_alg_type SMALLINT,
   deleted_flag STRING,
   action_cd STRING,
   workflow_run_id BIGINT,
   input_file_id BIGINT,
   session_inst_id INTEGER,
   info_system_id SMALLINT
   )
   STORED AS PARQUET
   LOCATION
   'hdfs:///data/custom/t/propagation/010/src/n010010002kul2_lmt_metric_hist'
   TBLPROPERTIES("external.table.purge"="true");

2. Запись данных Teradata в Hive‑таблицу:

   INSERT INTO custom_t_propagation_010_src.n010010002kul2_lmt_metric_hist@devsdpbdm_h_t_propagation
   (
       crncy_id,
       limit_id,
       limit_metric_amt,
       limit_metric_end_dt,
       limit_metric_start_dt,
       limit_metric_type_id,
       sod_alg_type,
       deleted_flag,
       action_cd,
       workflow_run_id,
       input_file_id,
       session_inst_id,
       info_system_id
   )
   SELECT
   crncy_id AS crncy_id,
   limit_id AS limit_id,
   CAST(limit_metric_amt AS DECIMAL(38,16)) AS limit_metric_amt,
   TO_CHAR(limit_metric_end_dt,'YYYY-MM-DD') AS limit_metric_end_dt,
   TO_CHAR(limit_metric_start_dt, 'YYYY-MM-DD HH24:MI:SS') AS limit_metric_start_dt,
   limit_metric_type_id AS limit_metric_type_id,
   sod_alg_type AS sod_alg_type,
   rtrim(deleted_flag) AS deleted_flag,
   rtrim(action_cd) AS action_cd,
   workflow_run_id AS workflow_run_id,
   input_file_id AS input_file_id,
   session_inst_id AS session_inst_id,
   info_system_id AS info_system_id
     from dev42_1_010_db_stg.n010010002kul2_lmt_metric_hist

3. Создание внешней таблицы в Greenplum для доступа к данным:

   DROP EXTERNAL TABLE IF EXISTS s_grnplm_as_t_didsd_010_db_stg_mg.n010010002kul2_lmt_metric_hist_ext;
   CREATE EXTERNAL TABLE s_grnplm_as_t_didsd_010_db_stg_mg.n010010002kul2_lmt_metric_hist_ext
   (
   crncy_id BIGINT,
   limit_id BIGINT,
   limit_metric_amt DECIMAL(38,16),
   limit_metric_end_dt VARCHAR(10),
   limit_metric_start_dt VARCHAR(32),
   limit_metric_type_id BIGINT,
   sod_alg_type SMALLINT,
   deleted_flag TEXT,
   action_cd TEXT,
   workflow_run_id BIGINT,
   input_file_id BIGINT,
   session_inst_id INTEGER,
   info_system_id SMALLINT
   )
   LOCATION ('pxf://custom_t_propagation_010_src.n010010002kul2_lmt_metric_hist?PROFILE=hive&SERVER=u_sklsdpbdm_s_custom_t_propagation')
       on all
   format 'custom' (formatter = 'pxfwritable_import');

4. Запись данных из Hadoop в Greenplum:

   INSERT INTO s_grnplm_as_t_didsd_010_db_stg.n010010002kul2_lmt_metric_hist
   (
       crncy_id,
       limit_id,
       limit_metric_amt,
       limit_metric_end_dt,
       limit_metric_start_dt,
       limit_metric_type_id,
       sod_alg_type,
       deleted_flag,
       action_cd,
       workflow_run_id,
       input_file_id,
       session_inst_id,
       info_system_id
   )
   SELECT
   crncy_id AS crncy_id,
   limit_id AS limit_id,
   limit_metric_amt AS limit_metric_amt,
   to_date(limit_metric_end_dt,'YYYY-MM-DD') AS limit_metric_end_dt,
   to_timestamp(limit_metric_start_dt, 'YYYY-MM-DD HH24:MI:SS') AS limit_metric_start_dt,
   limit_metric_type_id AS limit_metric_type_id,
   sod_alg_type AS sod_alg_type,
   deleted_flag AS deleted_flag,
   action_cd AS action_cd,
   workflow_run_id AS workflow_run_id,
   input_file_id AS input_file_id,
   session_inst_id AS session_inst_id,
   info_system_id AS info_system_id
   FROM s_grnplm_as_t_didsd_010_db_stg_mg.n010010002kul2_lmt_metric_hist_ext

Выполнение работ по переносу

После того, как выполнения всех подготовительных действия, мы начали перенос данных. Перенос данных был выполнен по следующим шагам:

Первый этап — создание всех необходимых таблиц. Мы запускали первый и третий набор скриптов, которые создавали таблицы в Hive для хранения данных на Hadoop и внешние таблицы в GreenPlum для доступа к данным на Hadoop.

Вторым этапом мы запускали вторые скрипты, которые копировали данные из Teradata в таблицы Hive на Hadoop. В этом нам помог, уже имеющийся у нас, собственный механизм параллельного запуска SQL команд с возможностью указания количества одновременно выполняемых потоков.

Третий этап — выполнение SQL‑скриптов на стороне Greenplum для записи данных из Hadoop в таблицы приемники на кластере Greenplum.

Перенос данных мы осуществляли в разрезе систем‑источников, которые загружались в наше хранилище. Если для некоторых систем мы понимали, что в период между загрузками данных перенести все данные не получится, то выбиралась часть статических данных, которая не меняется в инкрементальных загрузках, например, содержимое таблиц фактов за предыдущие дни и этот набор данных переносился заранее. Использование SQL-команд для выгрузки данных позволяло легко задавать любые необходимые условия фильтрации на данных, которые мы переносили.

Заключение

Мы успешно перенесли наш архив данных из Teradata в GreenPlum, используя наше решение, основанное на использовании промежуточного хранилища в Hadoop и фреймворка PXF. Опыт переноса показал, что за один день в период между ежедневными загрузками, мы полностью успевали переносить данные системы источника размером до 10 Тб с минимальным влиянием на текущую работу хранилища.