Как рассчитать DAU / MAU с использованием KSQL?

Question

Я разработчик мессенджеров и пытаюсь вычислить DAU / MAU, используя поток событий пользовательских запросов с использованием KSQL.

Я пытался вычислить его, используя следующий запрос:

CREATE TABLE ACTIVE_USER_ACTIONS_BY_1_HOUR WITH (
  KAFKA_TOPIC='active-user-actions-by-1-hour'
) AS
SELECT 
  MCCU.UID AS UID,
  COUNT(MCCU.UID) AS ACTIVITY_COUNT
FROM METRICS_REQUESTS MR
JOIN METRICS_CONTEXT_CID_UID MCCU ON MCCU.CID = MR.CID
WINDOW TUMBLING (SIZE 1 HOUR)
WHERE
  MR.REQ_NAME = 'SendMessage' OR
  MR.REQ_NAME = 'UpdateMessage'
GROUP BY MCCU.UID;

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

{
  "order": 3,
  "ROWTIME": 1570095657670,
  "ROWKEY": "1365010623 : Window{start=1570093200000 end=-}",
  "UID": 1365010623,
  "ACTIVITY_COUNT": 3
}
{
  "order": 1,
  "ROWTIME": 1570095651905,
  "ROWKEY": "1637035978 : Window{start=1570093200000 end=-}",
  "UID": 1637035978,
  "ACTIVITY_COUNT": 9
}

Не понимаю, как сопоставить эти строки чему-то вроде:

{
    "ACTIVE_UID_COUNT": 2,
    "START": 1570093200000,
    "END": null
}

Answer 1

Если вы хотите вычислить DAU / MAU, вы должны использовать функцию TOPKDISTINCT, которая подсчитывает только различные вхождения user_id во входящих событиях. Вы должны использовать SQL-код, как показано ниже:

CREATE TABLE dau_1min WITH (KAFKA_TOPIC='dau-1m', VALUE_FORMAT='AVRO') AS
SELECT
  country,
  WINDOWSTART() AS window_timestamp,
  ARRAYLENGTH(TOPKDISTINCT(user_id, 10000000)) AS dau
FROM eventssource
WINDOW TUMBLING (SIZE 1 MINUTE)
GROUP BY country;

Поскольку TOPKDISTINCT возвращает массив различных идентификаторов, вы должны реализовать пользовательскую функцию UDF, то есть ARRAYLENGTH, которая просто возвращает размер данного массива. После этого вы получите объект типа {country, window_timestamp, dau}, который достаточно прост для хранения в некоторой постоянной базе данных, такой как pqsql.