Пример 1. Работа с простыми файлами, преобразование, запись данных в базу. Автоматизация запуска графа.

Описание задачи

Разработать граф для загрузки данных из CSV файла в таблицу. Добавить поля со временем создания и обновления записей. Изменить в поле price символ точки, на запятую.

Настроить расписание и обработчик событий, вызвать созданный граф из другого графа.

Создание графа для преобразования данных и записи в базу

1. Чтобы создать новый граф, найдите в структуре своего проекта на панели Project structure папку grapf, в которую поместите граф. Кликните правой кнопкой на папку graph, выберите в меню пункт New grf file. Задайте имя графа, например load_ecommerce_behavior.grf и нажмите Создать.
2. Привяжите к графу внешний файл параметров: панель Outline -> Parameters -> Link parameters -> вставьте путь к файлу параметров ./workspace.prm в поле fileURL.
3. Привяжите также файл с описанием соединения с базой: панель Outline -> Connections -> Link connection -> вставьте путь к файлу соединения ./conn/demo2.con в поле dbConfig.
4. Перетащите на рабочую область с панели компонентов узлы CsvReader, Map, PostgresSqlDataWriter.
5. Соедините порты узлов рёбрами. Вытягивайте ребро из выходного порта (с правой стороны) узла и отпускайте, когда довели курсор до входного порта (с левой стороны) другого узла.
6. Создайте 2 меты через редактор метаданных.
   • на панели Outline откройте контекстное меню раздела Metadata -> выберите пункт New metadata. Задайте имя метаданных - ecommerce_behavior. Создайте 8 полей типа string: event_time, event_type, product_id, category_id, category_code, brand, price, user_id, user_session/ Чтобы добавить поля в мету используйте кнопку +. Чтобы удалить ненужное поле - кнопку -
   • на панели Outline откройте контекстное меню метадаты ecommerce_behavior -> выберите пункт Copy metadata. Скопируйте мету ecommerce_behavior и вставьте ее через контекстное меню раздела Metadata -> Paste metadata. Откройте редактор и смените название меты на ecommerce_behavior_out. Добавьте еще два поля типа date: created_at, updated_at
7. Перетащите мету ecommerce_behavior с панели Outline на ребро между узлами CsvReader и Map, а мету ecommerce_behavior_out на ребро между Map и PostgresSqlDataWriter.
8. Заполните значения атрибутов узлов. Чтобы открыть редактор узла, дважды кликните по его поверхности.
   • в CsvReader нужно вставить имя файла-источника данных в поле fileURL=${DATAIN_DIR}/2019-Nov.csv
   • в Map заполнить поле transform кодом трансформации на внутреннем языке Onebridge. Код приведён ниже в таблице "Атрибуты MAP".
   • в PostgresSqlDataWriter выберите имя нужного dbConnection из списка, вставьте имя таблицы для записи в поле table, и укажите параметры для записи в parameters

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

На рисунке ниже представлен созданный граф для загрузки данных из CSV файла в базу:

Атрибуты используемых узлов

Атрибуты CSV_READER:

Атрибуты MAP:

#[derive(Default)]
pub struct Local {
    pub start_date: DateTime,
}

fn transform(
    input: In, mut output: Out, 
    local: &mut Local, glb: Global,
) -> Result {
    if local.start_date == DateTime::NULL {
        local.start_date = DateTime::now();
    }

    output.0.event_time = input.0.event_time.clone();
    output.0.event_type = input.0.event_type.clone();
    output.0.product_id = input.0.product_id.clone();
    output.0.category_id = input.0.category_id.clone();
    output.0.category_code = input.0.category_code.clone();
    output.0.brand = input.0.brand.clone();
    output.0.price = input.0.price.replace(".", ",");
    output.0.user_id = input.0.user_id.clone();
    output.0.user_session = input.0.user_session.clone();
    output.0.created_at = local.start_date.clone();
    output.0.updated_at = local.start_date.clone();

    Result::All
}

Атрибуты POSTGRESQL_DATA_WRITER:

NULL=NULL

Вызов графа из другого графа

Чтобы вызвать созданный граф из другого графа, используйте узел ExecuteGraph.

1. Создайте новый граф, задайте имя, например startLoad
2. На рабочую область нужно поместить единственный компонент - ExecuteGraph
3. Откройте редактор атрибутов узла и задайте в поле jobURL имя графа, который нужно запустить: jobURL = ./graph/load_ecommerce_behavior.grf
4. Сохраните изменения атрибутов узла, сохраните граф CTRL+S.

Так будет выглядеть граф для вызова другого графа:

Теперь, если запустить startLoad.grf, он следом запустит внутренний граф load_ecommerce_behavior.grf.

Результат выполнения можно будет увидеть в панели администратора на странице История выполнения.

Общее время выполнения графа выводится в строку "Продолжительность" на вкладку Обзор. Затем можно перейти к просмотру данных по внутреннему графу через дерево графов в правой верхней части экрана.

Скорость выполнения внутреннего графа можно проверить на вкладке Отслеживание. В столбец "Длительность" выводится длительность работы каждого узла в рамках фазы. На приведенном ниже примере видно, что все узлы работали одновременно одинаковое количество времени.

В столбце "ЗАПИСЕЙ" указано общее количество записей, обработанных данным узлом. В столбце "ЗАПИСЕЙ/СЕК СР" - среднее количество записей, обрабатываемых узлом за секунду.

Рассчитаем общую среднюю скорость работы фазы: Общее количество записей поделим на общее время работы всех узлов, получится, что средняя скорость обработки около 207 698 записей в секунду. Для конкретного CSV-файла это примерно 27063 Кб в секунду.

Таким образом система может обработать 100Гб данных (около 800 млн записей) за ~65 минут.

Создание расписания

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

1. Для этого в панели администратора на странице Расписания диалог создания расписаний.
2. Задайте имя - Расписание запуска startLoad.grf, периодичность - Один раз, время исполнения, выберите проект и граф. Сохраните расписание.

Граф будет запущен в назначенное время.

Результат выполнения графа отобразится в панели администратора на странице История выполнения. На вкладке Обзор в поле "Тип запуска" будет указано по расписанию, а в "Пользователь" - root.

Создание обработчика событий

Если нужно, чтобы граф запускался по событию - нужно настроить обработчик для события. Триггерным событием для обработчика может быть окончание работы графа с определенным статусом либо создание/удаление файла в указанной директории.

1. В панели администратора, на странице Обработчики событий, вызовите диалог создания обработчика кнопкой Новый обработчик.
2. Задайте название обработчика, например, Поступил файл. Выберите запускающее событие - Файл и укажите необходимость проверить добавление файла в директорию /srv/projects/DEMO/data-in/*.csv. Задайте действие, которое надо выполнить - Запуск графа и его параметры: проект - DEMO, граф - /graph/startLoad.grf. Сохраните обработчик.

Когда триггерное событие случится, указанный в поле "Начало" граф будет запущен. Проверить выполнение графа можно на истории выполнения. На вкладке Обзор в поле "Пользователь" будет root, а в поле "Тип запуска" будет указано по какому событию запущен этот граф.