Вступление

Всем привет. Меня зовут Андрей, я работаю в компании SegmentStream, занимаюсь вопросами архитектуры, инфраструктуры и написанием кода. Сегодня я расскажу как о том, как была решена задача адаптации Cube под наши нужды. В рамках статьи я не буду подробно рассказывать про сам Cube, если вы не знакомы с Cube, то лучше сначала немного почитать про него на официальном сайте cube.dev/docs. Если говорить про SegmentStream, то мы разрабатываем SaaS-платформу для оптимизации рекламных кампаний на основе машинного обучения.

Также, в статье будет достаточное количество кода на TypeScript, который можно при желании не стесняться заимствовать. Я постараюсь оставлять объяснения к нему.

Зачем нужен Cube

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

Так как система работает с данными, то нужно иметь возможность эти данные как-то визуализировать. До этого клиенты могли смотреть отчёты через Google Data Studio, но проблема в том, что это был полностью сторонний инструмент, и даже его интеграция на сайт была через iframe.

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

Оценив все плюсы и минусы, было решено, что нужно переезжать на что-то своё, что-то, что будет стоить адекватных денег, и что будет закрывать все наши потребности. А исходные данные следующие:

Данные храняться в BigQuery, при этом, по большей части, в BigQuery клиента (данные-то клиенту принадлежат) и доступ выдаётся через OAuth2.0;
Данные, необходимые для отчётов, разбиты на 2 здоровенные таблицы по 20-30 миллионов строк с 10 различными ключами и 30 метриками каждая;
Обработка данных стоит денег в BigQuery, так что нужно, по возможности, избегать запросов в BigQuery;
Данные в этих двух таблицах обновляются по тригеру.

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

Компоненты следующие:

API отвечает за обработку входящих запросов;
Cube Refresh Worker отвечает за обновление данных по расписанию;
Redis используется как кеш и шина;
Кластер Cube Store отвечает за хранение преагрегаций и обработку запросов к ним.

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

Cube Store выгружает эти самые преагрегации из таблицы в файлы формата parquet и ходит в них во время запросов.

В общем, это всё очень понравилось и начался процесс интегрирацмм Cube.

Немного теории про Cube

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

В Cube данные описываются при помощи кубов (внезапно), куб представляет из себя описание того, как нужно извлекать данные из источника, например:

cube(`Users`, {
 sql: `SELECT * FROM users`,
 
 measures: {
   count: {
     sql: `id`,
     type: `count`,
   },
 },
 
 dimensions: {
   city: {
     sql: `city`,
     type: `string`,
   },
 
   companyName: {
     sql: `company_name`,
     type: `string`,
   },
 },
});

В коде выше мы определили базовый sql-запрос, 1 measure и 2 dimension. Если говорить простым языком, то measure это то, что нужно посчитать, а dimension - то, по чему нужно группировать. То есть, если нужно получить колличество пользователей в каждом городе, то нужно будет отправить в Cube следующий запрос:

{
  dimensions: ['Users.city'],
  measures: ['Users.count']
}

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

Преагрегаци определяются в кубе и выглядят как-то так:

cube(`Orders`, {
 sql: `SELECT * FROM users`,
 
 ...,
 
 preAggregations: {
    usersByCity: {
     dimensions: [CUBE.city],
     measures: [CUBE.count],
   },
 },
});

Так, как используется BigQuery, то имеет смысл немного рассказать про преагрегации в контексте работы с BigQuery. У Cube сущетсвует несколько стратегий создания преагрегаций:

Хранить в отдельной таблице в базе;
Стримом выгрузить эту отдельную таблицу из базы и отправить на хранение в Cube Store;
Использовать механизм exportBucket, при помощи которого BigQuery сам выгружает таблицу в файл и присылает в CubeStore ссылку на эти файлы.

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

Как Cube понимает, что пора обновлять данные? Для этого в преагрегацию необходимо добавить параметр refreshKey:

refreshKey: {
  every: `30 seconds`
  sql: `SELECT MAX(created_at) FROM users`,
}

Cube раз в 30 секунд будет выполнять этот запрос и при изменении результата будет пересчитывать преагрегации.

OAuth2.0

В Cube из коробки есть драйвер для BigQuery, вот только он умеет работать только с сервис-аккаунтами, а практически все клиенты сидят через OAuth 2.0 сидят. Что делать? Расширить драйвер, конечно же.

class SegmentStreamBigQueryDriver extends BigQueryDriver {
 config: BigQueryDriverOptions;
 constructor(config: BigQueryDriverOptions) {
   super(config);
 
   this.config = config;
   this.pinoLogger = pinoLogger;
 
   if (config.oauthCredentials) {
     const refreshClient = new UserRefreshClient();
 
     refreshClient.fromJSON({
       type: 'authorized_user',
       client_id: config.oauthCredentials.clientId,
       client_secret: config.oauthCredentials.clientSecret,
       refresh_token: config.oauthCredentials.refreshToken,
       project_id: config.projectId,
     });
 
     this.bigquery.authClient.cachedCredential = refreshClient;
   }
 }
}

Кстати, exportBucket работать с OAuth 2.0 не будет. Дело в том, что для подписи ссылок GCP использует ключ сервис-аккаунта, такие вот дела.

Авторизация пользователей

Теперь встаёт вопрос: как передать с фронта креды пользователя от BQ, которые нужно хранить и не отдавать никуда? Cube умеет в авторизацию через JWT, но вот только в данном случае нужно не просто проверить, что запрос пришёл от добросовестного пользователя, нужно ещё и креды от BigQuery передать.

После небольших раздумий было решено это всё засунуть JWT креды, предварительно зашифровав их. А через некоторое, время словив 431 Request Header, Fields Too Large ~~и побежали ставить костыли~~ пришлось добавлять флаги к ноде на увеличенный размер заголовков. Думаю, что сделать вы это и сами легко сможете, а интерфейс выглядит вот так:

interface EncodedCredentials {
 content: string;
 iv: string;
}
interface SecurityContext {
 projectId: string;
 datasetId: string;
 oauthCredentials?: EncodedCredentials;
 encodedServiceAccount?: EncodedCredentials;
}

Кто прочитал код, тот может заметить, что тут ещё и сервис-аккаунты поддержаны. Да, было решено, что нужно переводить пользователей на сервис-аккаунты и заканчивать гонять терабайты данных через инфраструктуру (BigQuery умеет экспортировать таблицу в файл в GCP, но как было сказано выше, умеет он это только с сервис-аккаунтом).

Итого, вся эта красота стала выглядеть следующим образом:

Фронт идёт в API и получает JWT для Cube, в JWT зашиты креды от BigQuery, у JWT, конечно же, есть подпись и срок жизни;
С полученным JWT и сформированным запросом фронт идёт в Cube;
Cube проверяет подпись JWT, расшифровывает креды и создаёт инстанс драйвера на основе SecurityContext;
Cube формирует ответ и отдаёт его клиенту.

В случае с Cube Refresh Worker (сервис, который следит, что все данные актуальны и обновляет преагрегаты) в API улетает запрос на те же данные, которые зашиты в JWT.

Преагрегации без поля updated_at

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

К счастью, ребята из команды по работе с данными вовремя подсказали один хак:

{
  granularity: `day`,
  partitionGranularity: `week`,
  buildRangeStart: { sql: `SELECT DATE_SUB(CURRENT_DATETIME(), INTERVAL 92 DAY)` },
  buildRangeEnd: { sql: `SELECT CURRENT_DATETIME()` },
  refreshKey: {
   every: `5 minutes`,
   sql: `SELECT COALESCE(max(last_modified_time), '2020-01-01T00:00:00.000Z') as last_updated_at
         FROM \`INFORMATION_SCHEMA.PARTITIONS\`
         where table_name = 'conversions'
           AND partition_id != '__NULL__'
           AND ${FILTER_PARAMS.Attribution.date.filter(
             "TIMESTAMP(PARSE_DATE('%Y%m%d', partition_id))",
           )}`,
  }
}

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

А через пару недель клиенты начали жаловаться на спам запросами… И это, честно говоря, было ожидаемо: каждые 5 минут Cube делал для каждой преагрегации и для каждой партиции запрос, после чего за сутки натекало неплохо так. А ещё BigQuery тратит 10 мегабайт на такой запрос (почему - большой вопрос), и за месяц может накапать под терабайт.

Итак, проблему нужно решать. Фактически, эти проверки вообще не нужны, и так прекрасно известно, когда была обновлена конкретная таблица (она же на бекенде и обновляеется), значит, нужно как-то ограничить походы Cube на BigQuery. Cube в бекенд с таким вопросом ходить не умеет, значит нужно пойти в Cube и подтюнить драйвер так, чтобы он умел кешировать запросы в redis, а кеш уже можно сбросить. Ниже приведён код того, что было сделалано. Заодно были замоканы проверки buildRangeStart и buildRangeEnd, потому что для этого в BigQuery лезть совсем не нужно.

 async runQueryJob<T = QueryRowsResponse>(
   initBigQueryQuery: Query,
   options: any,
   withResults = true,
 ): Promise<T> {
   // Parenthesized expression cannot be parsed as an expression, struct constructor, or subquery
   const bigQueryQuery = {
     ...initBigQueryQuery,
     query: this.replaceILIKE(initBigQueryQuery.query || ''), // BigQuery doesn't support ILIKE
   };
 
   if (bigQueryQuery.query === 'SELECT CURRENT_DATETIME()') {
     return [
       [
         {
           f0_: {
             value: new Date().toISOString(),
           },
         },
       ],
     ] as unknown as T;
   }
 
   if (bigQueryQuery.query === 'SELECT DATE_SUB(CURRENT_DATETIME(), INTERVAL 92 DAY)') {
     return [
       [
         {
           f0_: {
             value: new Date(Date.now() - 1000 * 60 * 60 * 24 * 92).toISOString(),
           },
         },
       ],
     ] as unknown as T;
   }
 
   if (this.cacheRenewewer && bigQueryQuery.query?.includes(`INFORMATION_SCHEMA.PARTITIONS`)) {
     const tableName = bigQueryQuery.query.match(/table_name = '([a-zA-Z0-9]*)'/);
     const { projectId, datasetId } = this.config;
     if (tableName && tableName[1] && projectId && datasetId) {
       const postfix = `${bigQueryQuery.params?.join('_').replace(/[-:\.]/, '_') || ''}_${
         tableName[1]
       }`;
       const cachedResult = await this.cacheRenewewer.getRecord(projectId, datasetId, postfix);
 
       if (cachedResult) {
         return cachedResult;
       }
 
       this.pinoLogger.info(
         {
           projectId,
           datasetId,
           postfix,
           tableName: tableName[1],
           params: bigQueryQuery.params,
         },
         'Request pre-aggregation last_updatet_at from DB',
       );
 
       const [preAggregationJob] = await this.bigquery
         .dataset(this.config.datasetId)
         .createQueryJob(bigQueryQuery);
       const result = (await this.waitForJobResult(
         preAggregationJob,
         options,
         withResults,
       )) as any as T;
       await this.cacheRenewewer.setRecord(projectId, datasetId, postfix, result);
       return result;
     }
   }
 
   const [job] = await this.bigquery.dataset(this.config.datasetId).createQueryJob(bigQueryQuery);
   return (await this.waitForJobResult(job, options, withResults)) as any as T;
 }

Код не самый тривиальный, так что лучше будет немного пояснить, что здесь вообще происходит: запросы на текущее время отдаются без походов в BigQuery (странно, что в Cube нельзя это сделать без костылей), все результаты ключа преагрегации кидаются в кеш и кеш их отдаёт, а после того, как на стороне бекенда обновили таблицы у пользователя, то кеш сбрасываем. Проблема решена, ура!

Утекающая память

Думаю, что многим знакомы вот такие картинки.

Этот раз тоже не стал исключением и мне пришлось наблюдать такие картинки после подключения Cube к крупным клиентам. Понятно, что я сам себе злой Буратино и нужно использовать сервис-аккаунты с exportBucket, но на тот момент у был только OAuth2.0. Да и, в конце концов, стримы не должны есть 15 гигабайт. После ~~console.log~~ дебага проблема была найдена, но уже не в Cube, а в драйвере BigQuery. Дело в том, что стриминг в драйвере BigQuery вообще не стриминг и ест память. Пришлось писать свою обёртку. Фактически, мы просто сделали стримы через пагинацию.

import { Readable } from 'stream';
import { BigQuery, Dataset, Job, QueryResultsOptions } from '@google-cloud/bigquery';
 
class BigQueryReadStream extends Readable {
 nextQuery: QueryResultsOptions | null;
 rowsBuffer: any[] = [];
 constructor(protected job: Job, protected limitPerPage: number) {
   super({
     objectMode: true,
   });
 
   this.nextQuery = {
     autoPaginate: false,
     maxResults: limitPerPage,
   };
 }
 
 protected manualPaginationCallback(
   err: Error | null,
   rows: any[] | null | undefined,
   nextQuery: QueryResultsOptions | null | undefined,
 ) {
   if (err) {
     return this.destroy(err);
   }
 
   this.nextQuery = nextQuery || null;
   this.rowsBuffer.push(...(rows || []));
   this.push(this.rowsBuffer.shift() || null);
 }
 
 protected readFromBQ() {
   if (!this.nextQuery) {
     return this.push(null);
   }
   this.job.getQueryResults(this.nextQuery, this.manualPaginationCallback.bind(this));
 }
 
 async _read() {
   if (this.rowsBuffer.length === 0) {
     this.readFromBQ();
   } else {
     this.push(this.rowsBuffer.shift());
   }
 }
}
 
interface BigQueryParams {
 bigquery: BigQuery;
 dataset?: string;
}
 
export const createQueryStream = async (
 bigquery: BigQueryParams,
 query: string,
 params: unknown[],
 limitPerPage = 1000,
): Promise<Readable> => {
 const instance: BigQuery | Dataset = bigquery.dataset
   ? bigquery.bigquery.dataset(bigquery.dataset)
   : bigquery.bigquery;
 
 // Run the query as a job
 const [job] = await instance.createQueryJob({
   query,
   params,
   parameterMode: 'positional',
   useLegacySql: false,
 });
 
 // For all options, see https://cloud.google.com/bigquery/docs/reference/v2/jobs/getQueryResults
 const queryResultsOptions = {
   // Retrieve zero resulting rows.
   maxResults: 0,
 };
 
 // Wait for the job to finish.
 await job.getQueryResults(queryResultsOptions);
 
 return new BigQueryReadStream(job, limitPerPage);
};

Ну и интегрировать написанное в сам Cube это было проще простого:

 public async stream(query: string, values: unknown[]): Promise<any> {
   const streamUUID = randomUUID();
   this.pinoLogger.info(
     {
       streamUUID,
       query,
       projectId: this.config.projectId,
       datasetId: this.config.datasetId,
     },
     `Starting stream ${streamUUID}: ${query}`,
   );
 
   const stream = await createQueryStream(
     {
       bigquery: this.bigquery,
       dataset: this.config.datasetId,
     },
     query,
     values,
     this.config.limitPerPage || 1000,
   );
   const rowStream = new HydrationStream({ streamUUID }, (message: string, params: any) =>
     this.pinoLogger.info(params, message),
   );
   stream.pipe(rowStream);
   return {
     rowStream,
   };
 }
}

Custom dimensions

Как вы помните, в рамках куба можно определить набор dimensions и преагрегаций, но что если пользователю нужно самому задавать dimensions? Схема-то для всех общая, а потребности у пользователей разные. Это было названо custom dimensions.

Для решения этой задачи в Cube тоже есть решение, но придётся чуть поработать напильником. Решение называется “Dynamic Schema Creation” и оно нам подходит, но есть нюанс.

Нюанс заключается в том, что security context используется только в момент компиляции схемы, соответственно, нужно связать security context и версию схемы:

function schemaVersion(context: Context) {
 const securityContext = extractSecurityContextFromContext(context);
 const customDimensions = (securityContext.customDimensions || [])
   .map(
     customDimension =>
       `${customDimension.key}:${customDimension.sql}:${customDimension.type || ''}`,
   )
   .join(':');
 const customDimensionsHash = crypto.createHash('sha256');
 customDimensionsHash.update(customDimensions);
 return `${getAppIdFromBQConfig(
   securityContext.projectId,
   securityContext.datasetId,
 )}__${customDimensionsHash.digest('hex')}`;
}

Отлично, теперь при изменении набора пользовательских dimensions версия схемы будет меняться, осталось только эти dimensions запихнуть в схему.

Сначала нужно извлечь их:

 const { securityContext } = COMPILE_CONTEXT;
 const customDimensions = securityContext.customDimensions || [];

А потом добавить в схему:

  dimensions: {    
    ...customDimensions.reduce((customDimensionsObject, customDimension) => {
       if (!customDimension.sql) {
         return customDimensionsObject;
       }
 
       customDimensionsObject[customDimension.key] = {
         sql: customDimension.sql.replace(/Report\.([a-zA-Z]+)/g, (_, dimension) => {
           return Report[dimension];
         }),
         type: customDimension.type || 'string',
       };
 
       return customDimensionsObject;
     }, {})
  }

Готово, теперь можно использовать custom dimensions при формировании запросов.

За границами преагрегации

Внезапно оказалось, что за границами преагрегации (вне buildRangeStart и buildRangeEnd) ничего нет, пустота, одни нули. Почему? А вот так, Cube видит, что есть преагрегация для запроса и идёт туда, не обращая внимания на то, что там есть не все данные.

Что делать? А ничего, если запрос выходит за границы преагрегации, то придётся как-то без преагрегации жить. Для этого нужно добавить фейковый measure:

notPreaggregatedMeasure: {
  sql: `0`,
  type: `number`,
}

На стороне клиента, если есть понимание, что запрос выходит за границы, то добавляем эту метрику, и запрос улетит в БД, ну или в кэш.

Сontinue wait

Иногда при запросе данных можно получить ошибку вида “продолжайте ждать, всё будет, но чуть позже”. В большинстве случаев это означает, что Cube что-то там строит на своей стороне.

Но проблема в том, что во-первых, это может затянуться, а во-вторых, если известно, что преагрегации строятся (очень хочется попросить команду Cube дать эндпоинт со статусом в разрезе security context) можно отдать пользователю запросы напрямую (через notPreaggregatetMeasure) и показать красивую крутилку о том, что что-то там строится.

В общем, этот ответ тоже можно использовать с пользой.

Кстати (вообще, некстати), к проблеме отсутствия эндпоинта со статусом. Как получить время обновления данных? Напрямую никак. Но можно чуть поковырять ответ и сделать вот так:

lastRefreshTime: new Date(
  Math.max(
    ...row
      .serialize()
      .loadResponse.results.map(result => new Date(result.lastRefreshTime).valueOf()),
  )
)

Шардинг Cube Worker

В самом начале я написал, что SegmentStream это SaaS. Значит количество SecurityContext, фактически, не ограничено, либо достаточно большое. Но Cube Worker не умеет обрабатывать часть SecurityContext, соответственно, могут появиться ощутимые задержки на обновление данных (export bucket есть далеко не у всех клиентов). Что делать? Шардировать. Как? Классическое консистентное хеширование.

Для консистетного хеширования нужно, чтобы каждый воркер знал свой номер и знал общее число воркеров. Для этого нам как нельзя лучше подходит StatefulSet в K8S. Нам нужно передать в каждый воркер его порядковый номер, шаблон имени и общее число реплик:

         env {
           name = "TOTAL_REPLICAS"
           value = local.replicas
         }
 
         env {
           name = "POD_GENERATED_NAME"
           value = "cubestore-workers-"
         }
 
         env {
           name = "POD_NAME"
           value_from {
             field_ref {
               field_path = "metadata.name"
             }
           }
         }

После чего при помощи библиоткеки hashring осуществить фильтрацию нужных контекстов:

export const podName = process.env.POD_NAME || 'cube-refresh-worker-0';
export const podGeneratedName = process.env.POD_POD_GENERATED_NAME_NAME || 'cube-refresh-worker-';
export const totalRepicas = Number(process.env.TOTAL_REPLICAS || 1);
 
const hashring = new HashRing(
 new Array(totalRepicas).fill(0).map((_, i) => `${podGeneratedName}${i}`),
);
 
const filterSecurityContext = (context: Context) => {
 const workerForContext = hashring.get(contextToAppId(context));
 return workerForContext === podName;
};
 
const apiClient = new ApiClient(process.env.API_URL || '');
 
const cube = {
 dbType: 'bigquery',
 externalDbType: 'cubestore',
 cacheAndQueueDriver: 'redis',
 scheduledRefreshContexts: async () => {
   const scheduledRefreshContexts = await apiClient.getAllSecurityContexts();
   const filteredContexts = scheduledRefreshContexts.map((bq: SecurityContext) => ({
     securityContext: bq,
   }))
 
   if (!filteredContexts) {
     pinoLogger.info(`Unable to fetch contexts`);
     return [];
   }
 
   pinoLogger.info(`Fetched ${filteredContexts.length} contexts`);
 
   const finalContexts = filteredContexts.filter(filterSecurityContext);
 
   return finalContexts;
 },
};

В результате, появляется возможность неограниченно горизонтально скейлить Cube Worker и существенно ускорить рассчёт преагрегаций.

SQL API

А напоследок, поговорим о самом вкусном - SQL API в Cube. Cube стабильно работал в продакшене: графики строились, таблицы рисовались. Но что опять не понравилось? Всё описанное выше отлично работает, но захотелось ещё больше гибкости. К сожалению, часть про формирование SQL на основе запросов тянет на отдельную статью, так что если интересно, пишите в комментариях, а здесь расскажу про основы.

Во-первых, внезапно, 50к строк и мегабайты данных гонять на фронт не очень нормально, нужно получать с бэкенда уже отсортированные данные с пагинацией.

Во-вторых, хочется иметь возможность получать с бэкенда кастомные метрики, например, разницу между какими-то двумя метриками или дифф по датам, при этом бекенд должен уметь всё это сортировать ещё по любому выбранному полю.

Через REST API делать такое нельзя. Там даже нельзя запросить разницу между двумя метриками, что, в принципе, и нормально, этот API для таких вещей и не предназначен.

Фактически, SQL API это API, которое совместимо с драйвером PostgreSQL и при помощи которого можно строить практически любые запросы. Приведу пример того, как формируется запрос:

const firstQuery = `SELECT
     ${[
    `${
        hasDimension ? dimension : "'none'"
    } as "${firstIntervalPrefix}dimension"`,
    ...selectMeasures,
    ...selectAdditionalMetrics,
].join(',\n')}
     FROM Report WHERE Report.date > '{{ DATE_FROM }}' AND  Report.date < '{{ DATE_TO }}' ${
    filter ? `AND ${converFilterIntoSQL(filter)}` : ''
}
     ${hasDimension ? `GROUP BY "${firstIntervalPrefix}dimension"` : ''}
   `;

if (!secondDateRange) {
    return firstQuery
        .replace(/\{\{ DATE_FROM \}\}/, firstDateRange.from)
        .replace(/\{\{ DATE_TO \}\}/, firstDateRange.to);
}

Сами метрики определяются подобным образом:

const reportAttributionFieldsMap = new Map([
 [ReportAttributionMetric.cost, 'MEASURE(Report.cost)'],
 [ReportAttributionMetric.clicks, 'MEASURE(Report.clicks)'],
 [ReportAttributionMetric.impressions, 'MEASURE(Report.impressions)'],
 [ReportAttributionMetric.users, 'MEASURE(Report.visitors)'],
 [ReportAttributionMetric.sessions, 'MEASURE(Report.visits)'],
 [ReportAttributionMetric.timeOnSite, 'MEASURE(Report.timeOnSite)'],
]);

Что на счёт additionalMetrics, так это фактически набор переменных и действий над этими переменными, где в качестве переменных выступают дефолтные метрики.

Не было бы лимита в 50к строк - было бы совсем хорошо.

А вот с получением метаданных возникла проблема. Как вы помните, у Cube нет эндпоинта для получения метаданных по преагрегациям, а из SQL API это не вытащить (по крайней мере в документации этого нет). Здесь пришлось использовать уже привычный REST API и обогащать результат полученными метаданными:

const restResult = await cubeJSClient
    .query(
        queries.dashboard.pingQuery.query([now, oneWeekAgo], conversionsIds),
    )
    .catch(() => undefined);

const sqlResult = await this.cubejsSQLAPI.requestTable(
    projectId,
    conversionsIds,
    {
        ...params,
        disablePreaggregation:
            params.disablePreaggregation ||
            restResult.buildingPreAggregations ||
            false,
    },
);

return {
    lastRefreshTime: restResult?.lastRefreshTime || new Date(),
    dimension: params.dimension,
    buildingPreAggregations: restResult?.buildingPreAggregations || false,
    rows: sqlResult.rows,
    totalRows: sqlResult.totalRows,
};

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

Ниже можно увидеть схему, как в итоге всё это работает.

Вместо заключения

Спасибо за то, что потратили время и прочитали эту статью, я надеюсь, что это было полезно. Уверен, что многое осталось за кадром, так что не стесняйтесь задавать любые вопросы в комментариях, с удовольствием на них отвечу.

Подводя итог можно сказать, что Cube очень удобный и полезный инструмент для построения отчётов и работы с данными. Да, в данном случае “из коробки” он работает не так хорошо, как хотелось бы, и, ещё много чего нужно доделать, но при желании это всё реализуемо, и использование Cube снимает очень много вопросов. И кстати, всё описанное с успехом работает как в k8s, так и в Cube Cloud.

Отдельное спасибо команде Cube за оперативные ответы, своевременные фиксы багов и ревью это статьи.

А вот такой красивый фронт получился под бекенд Cube SQL API. Можно увидеть, что есть возможность строить любые графики, сортировать по любым полям, сравнивать различые колонки, сортировать по полученному диффу, добавлять собственые dinmensions и раскрывать любые строки вглубь, а за всем этим стоит Cube SQL API.

Немного об интеграции Cube в SegmentStream