TensorFlow UDF

TensorFlow::InitSession(
  String, -- путь к сериализованному в protobuf формате графу вычислений, единственный обязательный аргумент
  Bool?, -- EnableMKLDNN, включение MKLDNN https://01.org/mkl-dnn
  Bool?, -- MKLCBWRCompatible, включение совместимости с SSE2 для MKL https://software.intel.com/en-us/mkl
  Bool?, -- MKLDNNUseWeightsAsGiven, по умолчанию false
  Bool?, -- AllowSoftPlacement, по умолчанию true
  Int32?, -- IntraOpParallelismThreads, по умолчанию 0
  Int32?, -- InterOpParallelismThreads, по умолчанию 0
) -> Resource<TensorFlow.Session> -- указатель на TensorFlow сессию с проинициализированным
                                  -- графом вычислений

TensorFlow::RunBatch(
  Resource<TensorFlow.Session>, -- результат TensorFlow::InitSession
  List<Struct<
    PassThrough:* -- произвольные данные, которые пропускаются через функцию без изменений
    *:* -- имена остальных полей соответствуют именам входов графа вычислений, которые
        -- будут переданы в TensorFlow, а их типы данных — соответствующим формам тензоров;
        -- примитивный тип данных соответствует скалярному тензору, список — вектору,
        -- матрицы на данный момент не поддерживаются.
  >>,
  Type<Struct<*:*>>, -- описание выходов графа, которые необходимо вычислить по аналогии со входами
  List<String>?, -- TargetNodeNames, опциональный список имен узлов графа, которые нужно вычислить,
                 -- но не возвращать
  Uint32?, -- BatchSize, по умолчанию 128
  Uint32?, -- MinVectorSize, по умолчанию 1, позволяет выроврять входные вектора до указанной длины,
           -- если на входе они имеют разную длину <= указанной.
           -- Если не выставлен, длины векторов батча выравниваются до максимальной в рамках батча.
) -> List<Struct<
  PassThrough:* -- переданное в аргумент PassThrough без изменений,
  *:* -- остальные поля соответствуют типу данных, указанному в третьем аргументе
>> -- результат

PRAGMA yt.DataSizePerJob = "128M"; -- поскольку это вычисление достаточно трудоемкое,
                                   -- будем запрашивать небольшой размер данных на задание
PRAGMA file("email.pb", "<путь к сериализованному графу вычислений>");
$session = TensorFlow::InitSession(
    FilePath("email.pb") -- путь к определению графа TensorFlow
);

PRAGMA library("char_table.sql"); -- считываем прикрепленный файл как источник для импорта
IMPORT char_table SYMBOLS $char_to_float; -- импортируем лямбду из библиотеки, которая преобразует символы
                                          -- электронной почты таким же образом, как это было сделано
                                          -- во время обучения модели

--лямбда-функция для предобработки адресов электронной почты
$email_to_float_vector = ($email) -> {
    $cut = FIND($email, "@"); -- оставляем только логин
    $cut = MIN_OF($cut, 20u) ?? 20u;
    $email = SUBSTRING($email, 0, $cut);  -- обрезаем слишком длинные логины
    $email_bytes = String::ToByteList($email); -- преобразуем строку в список байтов
    RETURN ListFlatMap($email_bytes, $char_to_float); -- применяем лямбду
};

$input = (
    SELECT
        $email_to_float_vector(email) AS input_1, -- имя столбца должно совпадать с именем
                                                  -- входа графа TensorFlow
        TableRow() AS PassThrough -- специальное имя столбца, чтобы передать
                                  -- любые данные в batch-режиме TensorFlow;
                                  -- в данном примере передается целая входная строка
    FROM emails_sample);
--------------------------------------------------------------------------------
-- Соглашения между TensorFlow и YQL:
-- - имя входного столбца должно совпадать с именем входного графа, а его тип должен
-- соответствовать форме тензора и типу данных;
-- - поддерживаемые типы данных: Float, Double, Int32, Int64, Uint8, Bool;
-- - List<> означает вектор, в то время как скаляр, матрицы и другие формы
-- пока не поддерживаются;
-- - у модели в этом примере только один вход, но их может быть много;
-- - те же правила применяются и к выходному графу;
-- - столбец PassThrough в настоящее время является обязательным, но может стать
-- опциональным в будущем.
--------------------------------------------------------------------------------

$processed = (PROCESS $input
USING TensorFlow::RunBatch( -- Apply TensorFlow graph in batches
    $session,     
    TableRows(), -- специальное имя для итератора по строкам таблицы
    -- описание вывода графа для вычисления
    Struct<network_output:List<Float>>,
    -- каждый элемент структуры представляет один из выходных данных графа;
    -- имя каждого элемента должно соответствовать имени, которое использовалось
    -- во время процесса обучения;
    -- соответствие между тензорами и типами данных YQL такое же, как и у входных данных
    ListCreate(String),   -- TargetNodeNames
    128,                  -- BatchSize
    20                    -- MinVectorSize
));

SELECT
    network_output[0] ?? 0.0f AS score, -- результат попытки угадать пол по электронной почте:
                                        -- - близко к 0.0 — "женский"
                                        -- - близко к 1.0 — "мужской"
                                        -- - близко к 0.5 — "трудно определить"
    PassThrough.email AS email, -- исходная электронная почта
    PassThrough.uid AS uid -- отдельный столбец для дополнительных входных данных
FROM $processed;