使用 Amazon Athena 做漏斗分析

在日常的业务运营过程中，管理者常常需要快速了解业务的运转健康状况，识别出瓶颈和问题，并制订应对的计划。要概览业务全景，我们可以借助一些统计工具。「漏斗分析」就是一种常见的工具，它很适合多个环节按时间顺序串联的业务。

现实生活中，很多业务都是多环节串联。比如电商用户可能会浏览推荐的物品、比价、加入购物车、下单购买、评价；再比如培训机构的客户可能会看到广告、咨询、试课、购买、正式上课、续费。

对于这类业务，我们可以把触达每个环节的人数统计出来，并形成一个逐渐递减的「漏斗」，就能看到每个环节的转化情况，定位到目前转化主要是卡在哪个步骤，再针对这个步骤补充定质的调研来确定解决方案。

在这篇文章中，我们将使用 Amazon Athena 来编写这样的一个漏斗分析工具，把一系列的时序数据（访问日志）转化成为每个环节的数量，再把数量转化成递减的漏斗。除了展示最终效果之外，我也会展示整个设计过程，帮助读者调整、设计自己的 SQL 语句。

Amazon Athena 介绍

Amazon Athena 是数据湖查询服务。它让用户可以使用 SQL 语句对存在 S3 上的半结构化数据（JSON、CSV、Parquet 等）进行查询。此外，它还是无服务器的服务，这意味着用户无需关心底层硬件资源，仅按照扫描数据的数量来进行收费。不扫描则没有其他闲置费用。

Amazon Athena 是基于 Presto 实现的。用户可以使用 Presto 的 SQL 语法和部分内置函数进行查询。

漏斗分析介绍

接下来我们来看漏斗分析。

在本文中，「漏斗分析」指的是：统计触达业务流程上每个关键环节的用户人数，并分析每个步骤的留存、转化、跳出率，以找到转化瓶颈。

漏斗分析包含的输入如下：

分析者关心的事件路径。比如我们关心「注册、浏览、下单」，那用户必须严格按照这个顺序来执行每个环节；例如：用户可以只执行「注册」，算走了 1 步，或者执行「注册、浏览」，算走了 2 步，但「注册、搜索、下单」只能算走了 1 步，因为「搜索」不在我们关心的路径内，从而打断了漏斗。
漏斗的时间区间。比如设置为 10 天，则漏斗包含的环节必须在 10 天内走完，如果用户第 1 天注册了，也浏览了，但是一直到第 20 天才下单，那么这个也只能算走了 1 步。

漏斗分析的输出有两个：

漏斗步骤计数（FUNNEL_COUNT）。比如 A 用户走了 1 步，B 用户走了 3 步，C 用户走了 2 步，等等。
漏斗人数统计。（FUNNEL_SUM）。比如走到第 1 步的有 1000 人，这其中走到第 2 步的有 300 人，而这其中走到第 3 步的又有 50 人，等等。

测试数据准备

我提前准备好了测试数据。数据结构如下：

数据字段如下。

event_name 是事件名字，中英文均可
user_id 是用户名
timestamp 是时间戳

数据是 JSON 格式，使用 Glue 爬虫进行爬取，录入为数据表，读者也可以自行创建外部表。如果手上没有现成数据，也可以使用 generatedata.com 等模拟数据生成工具来生成简单的测试数据。

漏斗语句设计过程

接下来我们来看如何一步步设计出漏斗分析语句。

简单分类统计

先来做步骤计数。先按用户来计算事件数量，这是一个分类统计的操作。既然是分类统计，我们直觉可能就会想到使用 GROUP BY 语句，比如：

SELECT user_id, COUNT(1) AS events_count
FROM events_table
GROUP BY user_id

这个语句的意思是，计算每个用户分别触发了多少事件。语句执行后，结果如下图：

限制时间窗口

这里有一个问题，就是它会统计这个用户所有触发过的事件，包括超出了我们设定的事件窗口范围的事件，所以，我们还需要做一次过滤。比如我们设定的时间区间是 5 天，那么下面示意图中 2020-11-23 以后的事件就必须被过滤掉。

需要注意的是，这个要求无法用简单的条件判断来解决，因为窗口的起始时间需要根据用户触发第一条事件的时间而定。比如用户 A 是 11 月 21 日触发第一条事件，那么窗口就从 11 月 21 日开始，而用户 B 是 12 月 1 日触发第一条事件，这个窗口则从 12 月 1 日开始。

要做到这一点，我们需要借助一个中间表。这个中间表只包含用户 ID 和这个用户的窗口结束时间，然后再用这个表来联结原来的事件表，从而可以用一个简单的条件判断来过滤掉超过时间窗口的数据。

包含窗口结束的中间表语句如下：

SELECT user_id,
  DATE_ADD('second', 3600 * 24 * 5, MIN(timestamp)) AS max_span
FROM events_table
GROUP BY user_id

我们首先使用 GROUP BY 按照把每个用户的事件分组，然后使用 MIN() 函数获取每个用户的第一条事件的时间，再使用 DATE_ADD() 函数，在这个时间的基础上增加了 5 天（假设窗口是 5 天），就得到了每个用户窗口结束的时间。

把这个表和简单统计的表联结起来，语句如下：

WITH max_spans_table AS
(
  SELECT user_id,
    DATE_ADD('second', 3600 * 24 * 5, MIN(timestamp)) AS max_span
  FROM events_table
  GROUP BY user_id
)
SELECT e.user_id, COUNT(1) AS event_count
FROM events_table e
JOIN max_spans_table m ON e.user_id = m.user_id
WHERE e.timestamp <= m.max_span
GROUP BY e.user_id

可以看出，这有点类似给第一个表做了一个遍历，并且给用户 ID 相同的条目加上了一个 max_span 参数，然后过滤掉了小于等于这个参数的行。

指定行为路径

不过这样还是有问题。虽然我们过滤了窗口外的数据，但并没有办法指定事件路径。我们关心的漏斗可能是「注册、浏览、下单」，但用户的行为可能是「搜索、浏览、下单」。我们还需要指定行为路径。

要在 SQL 语句里面对几条数据的顺序进行判断并不容易。因为 SQL 原本是用于操作和查询集合的，所以对顺序并不敏感。不过，还好事件名称只是简单的字符串，所以我们可以采用取巧的做法，把用户在某个时间窗口内的事件全部拼接成一个长的字符串，然后和我们预期的路径进行对比。

要把多条数据中的字段拼接到一起，我们需要两个函数。首先是 ARRAY_AGG()，用于把多条数据中的某个字段值取出来，组成一个数组。然后是 ARRAY_JOIN()，用于把数组中的字符串拼接到一起。

这里，我们假设我们关心的路径是「注册、询问客服、加入购物车、下单」。

此时的语句如下：

WITH max_spans_table AS
(
  SELECT user_id,
    DATE_ADD('second', 3600 * 24 * 5, MIN(timestamp)) AS max_span
  FROM events_table
  GROUP BY user_id
)
SELECT e.user_id, ARRAY_JOIN(ARRAY_AGG(e.event_name ORDER BY e.timestamp ASC), ',') AS event_seq, COUNT(1) AS event_count
FROM events_table e
JOIN max_spans_table m ON e.user_id = m.user_id
WHERE e.timestamp <= m.max_span
GROUP BY e.user_id
HAVING ARRAY_JOIN(ARRAY_AGG(e.event_name), ',') = 'REGISTER,INQUIRY,CART,ORDER'

注意虽然我们在 SELECT 时使用了别名 event_seq，但是给列取别名这个动作是在查询完成后在结果集中执行，所以 HAVING 语句中还是需要重复一遍 ARRAY_JOIN() 函数。此外，在 ARRAY_AGG() 函数中，我们还用 timestamp 字段做了排序，确保这些事件名字按照时间顺序拼接。

查询结果如下：

计算用户的漏斗路径长度

观察结果，会发现只有 7 个用户，而我们的测试数据集中有数千个用户，这明显不正常。究其原因，是因为我们只简单地比对拼接好的字符串和输入的行为路径，就只能查到全部路径都走完的用户。

比如，我们设定的输入是「A,B,C」，那么按顺序正好触发了这三个事件的用户就会拼接出「A,B,C」字符串从而比对成功。而只触发了 A，或者 A、B 的用户则无法匹配上。此外，A 事件如果不是第一个事件，而是在中间（「D,A,B,C」），又或者在目标事件路径后又触发了别的事件（「A,B,C,D」），简单的匹配都无法对比到。

此时我们需要做两件事：

找出某个用户的行为路径中第一个事件（比如「A」）所在的行，过滤掉它之前的行，否则路径在中间或者尾部就无法匹配（退一步说，即便可以匹配，我们也无法正确计算窗口时间）
只保留能以「A」「A,B」和「A,B,C」开头的事件路径，这是我们关心的目标路径

这时候我们就会触及 SQL 语言的表现力天花板。在普通编程语言中要做到这两件事很容易，因为我们可以多次循环，使用临时变量，再按需要过滤。而要在 SQL 中做到这样的过滤，就需要曲线救国了。

要做到第一点，我们需要把每个用户的事件单独提出来，删掉我们输入行为路径的第一个事件之前的其他事件。拿下图为例，如果我们关心的事件路径第 1 步是「注册」，那么在「注册」之前的步骤就都要删掉。

好在 SQL 中有提供「窗口函数」这样的概念。我们可以把每个用户的事件做成一个「窗口」，并且用 ROW_NUMBER() 函数给窗口中的每条数据一个行编号。接下来，我们要找到第一个事件所在的行编号，记录下来。然后，再联结原表，过滤掉这个用户下编号小于我们记录下的编号的行。这和前面的时间窗口过滤异曲同工。

窗口函数部分语句如下：

SELECT *,
  ROW_NUMBER() OVER (PARTITION BY id ORDER BY timestamp ASC) AS row_number
FROM events_table

使用 OVER () 语句，就形成了一个「窗口」，窗口由 PARTITION BY 指示的字段来作为划分，并且窗口内的数据条目可以排序。这有点像 GROUP BY 的逻辑，只不过 GROUP BY 只能用来做统计，而窗口函数则可以不做统计，只附加行编号。

执行结果如下：

可以看出，我们给每个用户自己的事件进行了编号，后续只需要查找到我们关心的事件所在的编号并联结原表做过滤即可。

过滤事件做到了，要做到第二点，多重匹配，则相对麻烦。

也许你会想到 LIKE ‘%%’ 等通配语句，但它只能匹配完整包含全部事件步骤的路径，无法匹配仅包含部分步骤的路径。如果使用 STRPOS()，再把两个参数反过来，那么确实可以匹配「A」「A,B」「A,B,C」以及「A,B,C,D,E」，但是却无法匹配「A,E,F,G」这样只以「A」开头的但后续是其他事件的路径。

幸好，我们还有 REDUCE() 函数，可以循环处理数组数据。只需先把行为路径字符串拆成一个数组，然后利用 REDUCE() 函数，使用递归的方式，把后一个字符串附加在前一个结果上，就能得到包含「A」「A,B」「A,B,C」等路径子集的数组，从而进行多重匹配。

要做到这一点，我们除了需要拿到数组中的字符串值，还需要拿到这个值所对应的下标。如果下标为 1 说明是第一个元素，原封不动，否则把当前的字符串取出来，和上一轮的字符串合并，添加到输出数组内。这个用 IF() 函数来做。

REDUCE() 函数本身没有提供下标参数，所以我们需要借助 CARDINALITY() 函数获取数组长度，再用 SEQUENCE() 函数来生成一个顺序数字的数组作为下标列表。遍历这个数组，就达到了类似其他语言的 for 循环或者 enumerate() 的效果。

这部分的语句如下：

SELECT REDUCE(
    SEQUENCE(1, CARDINALITY(event_steps)), ARRAY [], 
    (s, i) -> 
      IF(
        i > 1, 
        s || (CAST(s[i-1] AS VARCHAR) || ',' || event_steps[i]),
        s || event_steps[i]), 
    s -> s) AS event_step_combos
FROM (
  SELECT SPLIT('REGISTER,INQUIRY,CART,ORDER', ',') AS event_steps
)

这里用到了 CAST() 函数来把数组值设定为字符串，再使用 || 操作符来做字符串的连接。

下面是本轮的输出示例。因为控制台输出没有明确标明字符串，所以我把输出的部分用红线标记区分了一下。可以看出输出的数组中包含四个字符串，符合我们的要求。

不过，虽然有了这样一个数组，我们还需要对所有的事件路径做匹配，并且记录下匹配到的到底是「A」、「A,B」还是「A,B,C」，这样我们才能做后续的统计。要做到这一点，我们需要结合几个函数。

首先还是 REDUCE() 和 SEQUENCE() 函数，用于执行循环操作，最后输出一个数字，代表到底匹配到哪一个字符串。最后是 IF()，用于判断是否已经匹配到，如果已经匹配到，则维持原输入什么都不做，而如果没匹配到，则使用 STRPOS() 进行匹配。

因为一旦匹配出结果，就不会再进行匹配，所以我们必须先匹配最长的字符串。这就要求我们使用 REVERSE() 函数把字符串按照从长到短，反向排列。

这又引发一个问题，那就是因为我们的输入时数组反过来了，所以得出的数组下标也是反的。我们还必须再它下标反过来，也即是把下标 1、2、3、4 换成 4、3、2、1。通过分析可知，用数组长度减去下标再加 1 就可以把下标顺序反过来，所以我们把这部分添加上去。

把前面部分代码都融合到一起，此时代码如下：

WITH 
-- 拆分输入的事件路径
params AS
(
  SELECT REVERSE(REDUCE(
      SEQUENCE(1, CARDINALITY(event_steps)), ARRAY [], 
      (s, i) -> 
        IF(
          i > 1, 
          s || (CAST(s[i-1] AS VARCHAR) || ',' || event_steps[i]),
          s || event_steps[i]), 
      s -> s)) AS event_step_combos
  FROM (
    SELECT SPLIT('REGISTER,INQUIRY,CART,ORDER', ',') AS event_steps
  )
),
-- 给各个用户触发的事件进行独立行编号
events_with_row_numbers_table AS
(
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY timestamp ASC) AS row_number
  FROM events_table
),
-- 找到用户触发的漏斗中的第一个事件并记录行编号
events_starter_event_with_row_number AS (
  SELECT user_id, MAX(row_number) AS starter_event_row_number
  FROM events_with_row_numbers_table
  WHERE event_name = 'REGISTER'
  GROUP BY user_id
),
-- 过滤掉漏斗中第一个事件之前的事件
events_trimmed_table AS (
  SELECT e.user_id, e.event_name, e.timestamp
  FROM events_with_row_numbers_table e
  JOIN events_starter_event_with_row_number er ON e.user_id = er.user_id
  WHERE e.row_number >= er.starter_event_row_number
),
-- 找到时间窗口终点
max_spans_table AS
(
  SELECT user_id,
    DATE_ADD('second', 3600 * 24 * 5, MIN(timestamp)) AS max_span
  FROM events_trimmed_table
  GROUP BY user_id
),
-- 把用户的事件组合成一个字符串
events_seq_table AS
(
  SELECT e.user_id, 
    ARRAY_JOIN(ARRAY_DISTINCT(ARRAY_AGG(e.event_name ORDER BY e.timestamp ASC)), ',') AS event_seq
  FROM events_trimmed_table e
  JOIN max_spans_table m ON e.user_id = m.user_id
  WHERE e.timestamp <= m.max_span
  GROUP BY e.user_id
)
-- 找出用户停在了哪一步
SELECT e.user_id, e.event_seq, REDUCE(
  SEQUENCE(1, CARDINALITY(p.event_step_combos)), 
  0, 
  (s, i) -> IF(
    s > 0, 
    s, 
    IF(
      STRPOS(e.event_seq, p.event_step_combos[i]) = 1, 
      CARDINALITY(p.event_step_combos)-i+1, 
      0
    )
  ), 
  s -> s
) AS funnel_step
FROM events_seq_table e
CROSS JOIN params p

可以看出因为中间表的出现和函数的增加，查询语句已经变得很长了。我加入了注释，帮助读者看清楚每个中间表的目的。

此时结果如下：

可以看出，能匹配路径中的几个步骤，funnel_step 就是几。这样，我们就获得了每个用户走过的事件路径长度。

漏斗人数统计

接下来我们对每个步骤的人数做统计。

我们先计算每个步骤的用户数量，这是一个简单的统计。这部分代码如下：

SELECT COUNT(1) AS funnel_step_count
FROM events_funnel_step_table
GROUP BY funnel_step

可这个语句得出的结果是每个步骤单独的人数，但我们需要的不是单独的人数，而是漏斗，所以我们希望最后呈现的是环环相扣的统计。比如 1000 个人注册，其中 200 个人浏览，其中 15 个人下单。这意味着我们需要对步骤做累加。

这时候我们又需要借助窗口函数。窗口函数中有个特殊版的 SUM() 函数，这个版本的 SUM() 函数会把原来的「求和」变成「窗口间累加」，这符合我们的需要。此时我们省略了 PARTITION BY 语句，这就意味着每条记录自己就是一个窗口，而 SUM() 也变成了「按记录累加」的意思。

这部分代码如下：

SELECT 
  SUM(funnel_step_count) 
    OVER (ORDER BY funnel_step ASC) 
    AS funnel_step_acc_sum
FROM events_funnel_step_count_table

观察结果我们会发现，这和我们想的漏斗反过来了。因为漏斗是越来越少，而累加则是越来越多，我们需要用一个总数，逐步减去这个累加值。

这部分代码如下：

SELECT (funnel_total_sum-funnel_step_acc_count) AS funnel_step_converts
FROM events_funnel_step_acc_count_table
CROSS JOIN (
  SELECT SUM(funnel_step_count) AS funnel_total_sum
  FROM events_funnel_step_count_table
)

再次观察结果我们会发现，此时的漏斗最后会变成 0。这是因为我们的累加是从 0 开始，最后也会减到 0，而我们实际上希望只减到倒数第二步然后停止。换句话说，就是我们要用本行的值，去减上一行的累加值。

对于 SQL 来说，这就意味着一个表对自身做一个联结，但是要错开一行。用语句来表示如下：

SELECT e.funnel_step, (funnel_total_sum-e2.funnel_step_acc_count) AS funnel_step_converts
FROM events_funnel_step_acc_count_table e
LEFT JOIN events_funnel_step_acc_count_table e2 ON e.funnel_step = e2.funnel_step+1
CROSS JOIN (
  SELECT SUM(funnel_step_count) AS funnel_total_sum
  FROM events_funnel_step_count_table
)

结果如下。

观察结果又会发现一个问题，那就是第 1 行的值是空的。这是因为当左表是第 1 行时，右表的上一行不存在，所以 LEFT JOIN 的结果是 NULL，而针对 NULL 做数学运算，结果也只能是 NULL。

为解决这个问题，我们使用 COALESCE() 函数。这个函数可以帮助我们设置默认值。它会返回它的参数中第一个非 NULL 的值。我们把这个默认值设置为 0，这样，我们就可以对它进行正常计算。

加上后语句如下：

SELECT e.funnel_step, (funnel_total_sum-COALESCE(e2.funnel_step_acc_count, 0)) AS funnel_step_converts
FROM events_funnel_step_acc_count_table e
LEFT JOIN events_funnel_step_acc_count_table e2 ON e.funnel_step = e2.funnel_step+1
CROSS JOIN (
  SELECT SUM(funnel_step_count) AS funnel_total_sum
  FROM events_funnel_step_count_table
)
ORDER BY funnel_step ASC

效果如下：

参数提取

现在我们的目标就达成了。还剩下最后一个问题，那就是我们需要在不同语句，多次手动插入我们的参数。最好是我们能把参数都一次写到第一条语句中，使用起来更方便。

要做到这一点，我们使用 VALUES 来创建一个只有一行的临时表用于存储参数，然后用 AS 来给字段命名。使用时只需要从这个临时表 SELECT 需要的字段即可。

至此，我们用 Amazon Athena 的 SQL 编写了漏斗分析统计函数。因为 Amazon Athena 的底层是基于 Presto，所以这个语句也可以运行于兼容版本的 SQL 引擎上。

最终完整 SQL 语句如下：

WITH 
-- 原始参数
input AS
(
  SELECT time_window, SPLIT(event_chain, ',') AS event_steps
  FROM (VALUES ('REGISTER,INQUIRY,CART,ORDER', 3600 * 24 * 5))
  AS input (event_chain, time_window)
),
-- 拆分输入的事件路径
params AS
(
  SELECT time_window, event_steps, REVERSE(
    REDUCE(
      SEQUENCE(1, CARDINALITY(event_steps)), ARRAY [], 
      (s, i) -> 
        IF(
          i > 1, 
          s || (CAST(s[i-1] AS VARCHAR) || ',' || event_steps[i]),
          s || event_steps[i]), 
      s -> s
    )
  ) AS event_step_combos
  FROM input
),
-- 给各个用户触发的事件进行独立行编号
events_with_row_numbers_table AS
(
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY timestamp ASC) AS row_number
  FROM events_table
),
-- 找到用户触发的漏斗中的第一个事件并记录行编号
events_starter_event_with_row_number AS (
  SELECT user_id, MAX(row_number) AS starter_event_row_number
  FROM events_with_row_numbers_table
  WHERE event_name = (SELECT event_steps[1] FROM params)
  GROUP BY user_id
),
-- 过滤掉漏斗中第一个事件之前的事件
events_trimmed_table AS (
  SELECT e.user_id, e.event_name, e.timestamp
  FROM events_with_row_numbers_table e
  JOIN events_starter_event_with_row_number er ON e.user_id = er.user_id
  WHERE e.row_number >= er.starter_event_row_number
),
-- 找到时间窗口终点
max_spans_table AS
(
  SELECT user_id,
    DATE_ADD('second', (SELECT time_window FROM params), MIN(timestamp)) AS max_span
  FROM events_trimmed_table
  GROUP BY user_id
),
-- 把用户的事件组合成一个字符串
events_seq_table AS
(
  SELECT e.user_id, 
    ARRAY_JOIN(ARRAY_DISTINCT(ARRAY_AGG(e.event_name ORDER BY e.timestamp ASC)), ',') AS event_seq
  FROM events_trimmed_table e
  JOIN max_spans_table m ON e.user_id = m.user_id
  WHERE e.timestamp <= m.max_span
  GROUP BY e.user_id
),
-- 计算每个用户走完的事件路径长度
events_funnel_step_table AS
(
  SELECT e.user_id, e.event_seq, REDUCE(
    SEQUENCE(1, CARDINALITY(p.event_step_combos)), 
    0, 
    (s, i) -> IF(
      s > 0, 
      s, 
      IF(
        STRPOS(e.event_seq, p.event_step_combos[i]) = 1, 
        CARDINALITY(p.event_step_combos)-i+1, 
        0
      )
    ), 
    s -> s
  ) AS funnel_step
  FROM events_seq_table e
  CROSS JOIN params p
),
-- 计算走完不同事件路径长度的人数
events_funnel_step_count_table AS
(
  SELECT funnel_step, COUNT(1) AS funnel_step_count
  FROM events_funnel_step_table
  GROUP BY funnel_step
),
-- 按顺序累加事件路径上每个步骤的人数
events_funnel_step_acc_count_table AS
(
  SELECT funnel_step, SUM(funnel_step_count) OVER (ORDER BY funnel_step ASC) AS funnel_step_acc_count
  FROM events_funnel_step_count_table
),
-- 给用户事件路径长度统计增加一个常用的别名
funnel_count AS (
  SELECT * FROM events_funnel_step_count_table
),
-- 把累加变成错一行累减获得步骤转化
funnel_sum AS (
  SELECT e.funnel_step, (funnel_total_sum-COALESCE(e2.funnel_step_acc_count, 0)) AS funnel_step_converts
  FROM events_funnel_step_acc_count_table e
  LEFT JOIN events_funnel_step_acc_count_table e2 ON e.funnel_step = e2.funnel_step+1
  CROSS JOIN (
    SELECT MAX(funnel_step_acc_count) AS funnel_total_sum
    FROM events_funnel_step_acc_count_table
  )
  ORDER BY funnel_step ASC
)
SELECT *
FROM funnel_sum

每个中间表都可以单独 SELECT 出来作为调试优化之用。最后的 funnel_count 和 funnel_sum 表则可以用于做漏斗分析。

总结

本文介绍了如何使用 Amazon Athena 来进行漏斗分析，并重点介绍了完整的思考过程和中间用到的 SQL 语法及函数。希望能帮助读者对数据湖进行更高效的探索。

亚马逊AWS官方博客