Для преобразования данных

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

Список шагов для преобразования данных:

Sort - сортирует записи

Filter - фильтрует записи

Gather - собирает записи из разныых потоков в один

Copy - копирует записи

Concat - объединяет записи

Map - пользовательский алгоритм обработки

Dedup - устраняет копии по ключу

Normalizer - преобразует запись с несколькими полями, разбивая её на несколько записей с одним полем (нормальзует запись)

DataIntersection - объединяет отсортированные потоки данных, может применять преобразование

Sort

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

Атрибуты Sort:

Пример. Сортировка данных.

Входные записи содержат имена файлов и их размер. Нужно отсортировать файлы по размеру, начиная с самого большого (descending – по убыванию). Метаданные содержат поля «FileName», «FileSize».

Входящие записи:

Решение:

Ключ сортировки: FileSortKey="FileSize(d)"

Исходящие записи:

Filter

Filter фильтрует входные данные в соответствии с логическим выражением. Отправляет все записи, соответствующие выражению фильтра, в первый выходной порт и все отклоненные записи во второй выходной порт.

<attr name="filterExpression">
    <![CDATA[ 
        $in[0].count != 177 && $in[0].product == «карандаш» 
    ]]>
</attr>

Пример. Фильтрация данных.

Входные данные содержат данные о продуктах, проданных в прошлом году. Нужно узнать данные только по карандашам. Метаданные содержат поля Product, Count и Location.

Входящие записи:

Решение:

Выражение для фильтрации: $in[0].product == «карандаш»

Исходящие записи:

Gather

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

Пример. Сбор записей с нскольких входных портов.

Нужно собрать записи с нескольких входов. Метаданные содержат одно поле «id».

Входящие записи:

Решение:

Описание метаданных:

    <Global>
        <Metadata id="ObjectWithPos">
                <Field name="id" type="int"/>
        </Metadata>
    </Global>

Объявление шага:

<Node id="x" guiX="250" guiY="220" guiName="Gather" type="Gather"/>

Исходящие записи:

Copy

Copy получает записи через один входной порт и копирует каждую из них на все подключенные выходные порты. Шаг не имеет атрибутов.

Пример. Копирование данных.

Нужно скопировать записи с метаданными «carID» и «mark» в три потока.

Входящие записи:

порт 0:

Решение:

Для копирования в несколько потоков нужно подключить Copy несколько выходных портов. Записи на всех выходных портах будут идти в одинаковом порядке.

Исходящие записи:

порт 0:

порт 1:

порт 2:

Concat

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

Пример. Объединение записей.

Нужно объединить записи. Поданные на вход метаданные имеют поля «flower», «color».

Входящие записи:

порт 0:

порт 1:

порт 2:

Решение:

После конкатенации будут получены следующие записи.

Исходящие записи:

порт 0:

Map

Map позволяет написать пользовательский алгоритм обработки данных, используя внутренний язык системы. Можно по своему усмотрению трансформировать данные между входным и выходными портами, если предложенных шагов не хватает для выполнения необходимых преобразований данных.

Имеет единственный входной порт и как минимум один выходной. Может отправлять разные записи в разные выходные порты или даже отправлять одну и ту же запись на несколько выходных портов. Работает только с одним элементом, сохраняет порядок записей.

С помощью Map можно:

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

<attr name="transform">
    <![CDATA[
        function transform() {
            $out[0].person = $in[0].name.toString() + "_" + $in[0].surname.toString();
            $out[1].person = $in[0].name.toString().toUpperCase() + " " + $in[0].surname.toString().toUpperCase();
            return ALL;
        }
    ]]>
</attr>

Пример. Обработка данных с помощью Map.

Нужно получить произведение и сумму полученных на вход данных и отправить результаты на разные выходные порты. Входные метаданные содержат поля a, b. Нужно отправить результат перемножения a*b на первый порт, а результат сложения a+b на второй порт.

Входящие записи:

Решение:

Преобразование:

<Attr name="transform"><![CDATA[
    pub fn transform() -> OutPort {
        let res_mul = $in[0].a * $in[0].b;
        let res_add = $in[0].a + $in[0].b;

        $out[0].res_mul = res_mul;
        $out[1].res_add = res_add;

        return ALL;
    }

    ]]>
</Attr>

Исходящие записи:

порт 0:

порт 1:

Dedup

Dedup удаляет повторяющиеся записи по ключу.

Порты Dedup:

Атрибуты Dedup:

Сортированный и несортированный ввод

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

Если ваши данные содержат несколько разных значений ключей, вы можете использовать несортированный ввод.

Дедуплицирование несортированных входных данных не требует предварительной сортировки, но ограничивается доступной памятью, поскольку записи, которые должны быть отправлены на первый выходной порт, кэшируются. Требования к оперативной памяти в несортированном режиме зависят от значений атрибута dupAmount. Меньшее количество дубликатов означает, что требуется меньше памяти.

Если вы используете несортированные входные записи, порядок выходных записей не обязательно будет таким же, как порядок входных записей.

Пример. Дедупликация сортированных записей.

Записи содержат время входов на некоторый ресурс с различных ip адресов. Записи отсортированы по времени входа. Нужно найти время первого входа для каждого ip адреса. Метаданные содержат поля «ip» и «time».

Входящие записи:

Решение:

Ключ дедупликации: dedupKey = «ip»

Исходящие записи:

Normalizer

Normalizer создает одну или несколько выходных записей из каждой отдельной входной записи. Входные записи не обязательно сортировать.

Порты Normalizer:

<attr name="normalize"><![CDATA[
    function count() {
	n++;

	return length($in[0]);
    }
    function transform(integer idx) {
	$out[0].rn = n;
	$out[0].name = getFieldLabel($in[0], idx);
	$out[0].value = getStringValue($in[0], idx);
	
	return ALL;
    }
]]></attr>

Для Normalizer нужно определить функцию преобразования. Преобразование должно быть определено в файле задания в артибуте normalize. Эта функция будет вызываться заданное количество раз для каждой записи, поданной на вход этому шагу.

Порядок вызова функций описан ниже.

Алгоритм нормализации записей

Функции шага Normalizer:

function integer count() {
   customers = split($in.0.customers,";");
   return length(customers);
}

function integer transform(integer idx) {
   myString = customers[idx];
   $out.0.OneCustomer = str2integer(myString);
   $out.0.RecordNo = $in.0.recordNo;
   $out.0.OrderWithinRecord = idx;

   return OK;
}

Пример. Преобразование записи с многозначными полями в несколько записей.

Входные записи содержат название должности и список имён сотрудников. Нужно преобразовать записи в кортежи, содержащие название должности и одно имя сотрудника.

менеджер | [Егор, Алина]
разработчик | [Артём, Никита, Данил]

Решение

Определим преобразование, используя атрибуты Normalize:

function integer count() {
    return length($in.0.users);
}

function integer transform(integer idx) {
    $out.0.group = $in.0.group;
    $out.0.user = $in.0.users[idx];
    return OK;
}

Normalizer вернёт следующие записи:

менеджер |Егор
менеджер |Алина
разработчик|Артём
разработчик|Никита
разработчик|Данил

Заметка: Если преобразование указано во внешнем файле, рекомендуется явно указать исходную кодировку в артибуте charset.

DataIntersection

DataIntersection получает отсортированные данные с двух портов, сравнивает значения ключей в обоих и обрабатывает записи следующим образом:

• Такие входные записи, которые находятся как на входном порте 0, так и на входном порте 1, обрабатываются в соответствии с определяемым пользователем преобразованием, и результат отправляется на выходной порт 1.

• Такие входные записи, которые находятся только на входном порте 0, отправляются без изменений на выходной порт 0.

• Такие входные записи, которые есть только на входном порте 1, отправляются без изменений на выходной порт 2.

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

Преобразование должно быть определено, если подключён выходной порт 1.

Ключ соединения

Выражается как последовательность отдельных подвыражений, отделенных друг от друга точкой с запятой. Каждое подвыражение представляет собой присвоение имени поля из первого входного порта (с префиксом в виде знака доллара $) слева и имени поля из второго входного порта (с префиксом $) с правой стороны.

Заметка: Компонент может возвращать количество записей, отличное от исходного количества входных записей.
Если для параметра keyDuplicates установлено значение false, количество выходных записей может быть меньше количества входных записей, поскольку используется только первая из записей с дубликатом ключа.
Если для параметра keyDuplicates установлено значение true, количество выходных записей может быть больше, чем количество входных записей. На выходе создается декартово произведение записей, имеющих одинаковый ключ.