离线分析计算的解决方案调研

背景

随着数据量日益庞大，统计计算的需求也愈加“多样化”，伴随着连表、嵌套子查询，然后这类“不友好” SQL 在生产环境下是被禁止执行的；还有些情况需要对表字段二次加工再进行逻辑处理，对于这类需求，一般是把数据抓取下来通过程序做计算，但数据量庞大的情况下这种方式肯定不可取，极有可能会造成 OOM。面对复杂的统计需求，传统型关系数据库下直接查询或通过应用服务器进行二次加工计算的方式显得不可取，本文旨在介绍阿里云的相关解决方案。

方案简述

RDS 分析实例

优势：与 RDS 深度集成，内置 DTS，不需考虑手动操作 DTS 进行数据同步。
劣势：不支持 PolarDB；它会自动同步主实例的所有数据，相对而言灵活性差、经济成本高。

云原生数据仓库 ADB

优势：支持 PolarDB；使用量灵活扩容。
劣势：经济成本略高，需手动操作 DTS 进行同步。

可以看作是 MySQL 的升级版，主要作用是对 SQL 查询加速。生产环境下的业务数据库一般禁止不友好（重计算、耗时长）的 SQL 运行，这是可以使用 ADB 来解决长难复杂 SQL 的执行。

ADB 的使用步骤：

  1. 使用 DTS 同步工具将业务数据表的数据导入到 ADB 数据库下的相应数据表下；
  2. 执行目标 SQL 查询结果。

MaxCompute

优势：大幅降低业务数据库及在线应用服务器的计算压力；支持自定义函数来满足不同的计算需求；支持多种类型数据库。
劣势：有一定的学习成本；经济成本略高。

可以看作为一种计算能力，最重要的是它支持自定义函数，我们可以通过编码封装自己的算法来实现复杂的统计计算需求。举个例子，现在有一张数据量很庞大的 MySql 数据表，下面给出它在 MaxCompute 下的表结构定义：

user_authorization_info_base（用户授权信息表）

属性名称 类型 可空（Y/N） 描述
id BIGINT N 记录ID
user_id BIGINT N 用户ID
profession STRING N 用户行业信息
authorization_detail STRING Y 用户授权详情（json格式）

其中 authorization_detail 属性的内容格式为 json 文本，内容示例：

1
2
3
4
5
6
7
8
9
10
11
12
{
    "authorizationType": 5,
    "countryCode": "+86",
    "creditCode": "912345677912345212",
    "enableCloseLimit": 5,
    "enableUpdateLimit": 1,
    "enterpriseAddress": "北京市朝阳区***",
    "enterpriseEmail": "uname@**",
    "enterpriseName": "北京**科技有限公司",
    "enterprisePhone": "178****2357",
    "enterpriseScale": 0
}

现在有一个需求：获取填写了授权信息（即 authorization_detail 不为空）且授权类型为单次授权或永久授权（即 authorizationType 取值 5 或 10）的用户的 ID、授权手机号（enterprisePhone）、授权详情（authorization_detail）。通过关系型数据库下的 SQL 查询很难实现该需求，但借助 MaxCompute 实现起来会很简单，本文后续会以该场景给出一个 MaxCompute 的使用 Demo。

MaxCompute 的使用步骤：

  1. 使用 DTS 同步工具将业务数据表的数据导入到 MaxCompute 下的相应数据表下；
  2. 项目中使用相应 SDK 调用 MaxCompute 的资源进行计算。

MaxCompute 使用示例

下面以解决上述需求给出 MaxCompute 的简单使用示例。

IDEA 安装 MaxCompute 插件并连接项目

连接成功后可看到如下界面：

创建数据表并导入数据

步骤略。

开发 UDF

UDF 全称为 User Defined Function，即用户自定义函数。MaxCompute 提供了很多内建函数来满足计算需求，同时还可以通过创建自定义函数来满足不同的计算需求。UDF 在使用上与普通的内建函数类似，Java 和 MaxCompute 的数据类型的对应关系，可参见参数与返回值类型。

MaxCompute 支持的 UDF 有三种：

User Defined Scalar Function（通常也称之为 UDF）
用户自定义标量值函数（User Defined Scalar Function）。其输入与输出是一对一的关系，即读入一行数据，写出一条输出值。

UDTF（User Defined Table Valued Function）
自定义表值函数，是用来解决一次函数调用输出多行数据场景的，也是唯一能返回多个字段的自定义函数。而 UDF 只能一次计算输出一条返回值。

UDAF（User Defined Aggregation Function）
自定义聚合函数，其输入与输出是多对一的关系， 即将多条输入记录聚合成一条输出值。可以与 SQL 中的 Group By 语句联用。具体语法请参见聚合函数。

我们需要输出多行数据，则应该使用 UDTF。方法接收到 authorization_detail 属性值，判断非空再提取授权类型二次判断，符合条件则输出授权手机号，完整的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.UDFException;
import com.aliyun.odps.udf.UDTF;
import com.aliyun.odps.udf.annotation.Resolve;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;

import java.util.Arrays;

/**
 * MaxCompute 自定义函数，其功能如下：
 * 获取填写了授权信息（即 authorization_detail 不为空）且授权类型为单次授权或永久授权（即 authorizationType 取值 5 或 10）
 *   的用户的 ID、授权手机号（enterprisePhone）、授权详情（authorization_detail）
 */
// define input and output types, e.g. "string,string->string,bigint".
@Resolve({"bigint,string->bigint,string,string,smallint"})
public class AuthUserUDTF extends UDTF {

    /**
     * 目标授权类型
     */
    private static final int[] TARGET_TYPES = new int[]{5, 10};

    /**
     * 初始化方法，在 UDTF 处理输入数据前，调用用户自定义的初始化行为。在每个 Worker 内 setup 会被先调用一次。
     * @param ctx 执行程序上下文
     * @throws UDFException 操作错误
     */
    @Override
    public void setup(ExecutionContext ctx) throws UDFException {

    }

    /**
     * 这个方法由框架调用，SQL 中每一条记录都会对应调用一次 process，process 的参数为 SQL 语句中指定的 UDTF 输入参数。
     * 输入参数以 Object[] 的形式传入，输出结果通过 forward 函数输出。需要在 process 函数内自行调用 forward，以决定输出数据。
     * @param args 传入参数信息
     * @throws UDFException 操作错误
     */
    @Override
    public void process(Object[] args) throws UDFException {
        Long userId = (Long) args[0];
        String phone = "";
        String authorizationDetail = args[1] != null ? (String) args[1] : "";
        boolean matched = false;
        if (!"".equals(authorizationDetail)) {
            // 示例 json 如下：
            //    {"authorizationType":5,"countryCode":"+86","creditCode":"912345677912345212","enableCloseLimit":5,"enableUpdateLimit":1,"enterpriseAddress":"北京市朝阳区***","enterpriseEmail":"uname@**","enterpriseName":"北京**科技有限公司","enterprisePhone":"178****2357","enterpriseScale":0}
            /*
                不能通过fastjson完成数据解析，因为json、protobuf 类的包经常免不了要用反射的，会被沙箱挡住。
                解决方案是使用ODPS内置的gson完成json数据解析。
             */
            try {
                JsonObject jo = new JsonParser().parse(authorizationDetail).getAsJsonObject();
                if (matched = jo.has("authorizationType")
                        && Arrays.binarySearch(TARGET_TYPES, jo.get("authorizationType").getAsInt()) >= 0) {
                    phone = jo.get("enterprisePhone").getAsString();
                }
            } catch (Exception ignored) {
            }
        }
        Short match = (short) (matched ? 1 : 0);
        // 调用 forward 方法输出数据，每次 forward 代表输出一条记录。对应 SQL 语句 UDTF 的 as 子句指定的列。
        forward(userId, phone, authorizationDetail, match);
    }

    /**
     * UDTF 的结束方法，此方法由框架调用，并且只会被调用一次，即在处理完最后一条记录之后。
     * @throws UDFException 操作错误
     */
    @Override
    public void close() throws UDFException {

    }

}

打包项目为 Jar 并上传到 MaxCompute

打包过程省略，上传操作如下：

注册方法

操作方法如下图：

执行 SQL 获取目标结果

SQL 脚本如下：

1
2
3
4
5
6
SELECT user_id, phone, authorization_detail
FROM
(SELECT auth_user(user_id, authorization_detail) AS (user_id, phone, authorization_detail, match)
FROM user_authorization_info_base
) t
WHERE t.match = 1;

子查询中我们调用了自定义的函数 auth_user()，传入了 user_id/authorization_detail 等属性，经过函数的处理后会输出 user_id/phone/authorization_detail/match 等结果，match 表示是否满足筛选条件（1 为满足，0 为不满足），在最外层的 WHERE 条件后限制了 match = 1。
执行 SQL，得到结果如下：

需要注意的是，上述 SQL 执行结果最多返回 10000 条记录，在此只做一个展示为了验证能实现需求，业务代码中需结合 Tunnel 导出全量结果。

相关文档

RDS 分析实例简介
ADB MySQL 简介
MaxCompute 简介
MaxCompute UDF 开发指南
SQLTask 配合 Tunnel 实现大量数据导出

大数据
#adb #maxcompute
离线分析计算的解决方案调研
https://blog.yohlj.cn/posts/fdb72ad/
作者
Enoch
发布于
2021年1月20日
许可协议