自定义Join（UDJ） - 云原生大数据计算服务 MaxCompute

基于MaxCompute 2.0计算引擎，MaxCompute在UDF框架中引入了新的扩展机制UDJ（User Defined Join），来实现灵活的跨表、多表自定义操作，同时减少通过MapReduce等方式对分布式系统底层细节的操作。

背景信息

MaxCompute内置了多种Join操作，包括Inner/Right Join、Outer/Left Join、Outer/Full Join、Outer/Semi/Anti-semi Join等。这些内置的Join操作功能强大，但由于其标准的Join实现，无法满足很多跨表操作的需求场景。

通常，您可以通过UDF（User Defined Function）描述代码框架，但现有的UDF/UDTF/UDAF接口主要是针对在单个数据表上的操作而设计。一旦涉及多表的自定义操作，经常还需要依赖于内置Join +各种UDF/UDTF，并配合比较复杂的SQL语句来完成。在多表操作的场景上，您不得不放弃SQL而使用自定义MapReduce，才能完成所需的计算。

无论是Join +各种UDF/UDTF+复杂SQL还是自定义MapReduce门槛都比较高，同时还会带来一些问题：

使用Join +各种UDF/UDTF+复杂SQL：多个复杂的Join和散布在SQL语言各处的代码揉合在一起，将带来多处的逻辑黑盒，不利于生成最优的执行计划。
使用MapReduce：不仅更大程度上限制了系统进行执行优化的可能性，而且在深度优化本地运行代码时，由于MapReduce绝大部分代码由Java完成，在执行效率上会远低于MaxCompute基于LLVM的代码生成器。

使用限制

目前版本不支持使用UDF/UDAF/UDTF读取以下场景的表数据：

做过表结构修改（Schema Evolution）的表数据。
包含复杂数据类型的表数据。
包含JSON数据类型的表数据。
Transactional表的表数据。

UDJ的性能

通过一个真实的线上MapReduce作业进行测试验证UDJ的性能，该MapReduce作业实现一套比较复杂的算法。将两个表合并在一起，用UDJ对该MapReduce进行改写，并且验证UDJ实现结果的正确性。在并发度相同的情况下，两者性能对比如下。

由上图可见，UDJ接口的引入，一方面让您能更方便地描述对多表数据进行操作的复杂逻辑，一方面大幅提升了性能。代码只在UDJ内被调用，其上下游的逻辑（例如该示例中的整个Mapper逻辑）则完全通过MaxCompute高效的Native运行完成。在Java代码中，由于MaxCompute UDJ运行引擎和Java接口之间的数据交互逻辑有深度的优化，通过UDJ实现的Join逻辑也比其对等的Reducer更高效。

UDJ跨表Join功能

通过如下样例为您详细介绍MaxCompute UDJ跨表Join的使用方法。

假设存在两个日志表，分别是payment和user_client_log。

payment：表中保存了用户的支付记录，每一笔支付记录包含用户ID、支付时间和支付内容。样例数据如下。
user_id
time
pay_info
2656199
2018-02-13 22:30:00
gZhvdySOQb
8881237
2018-02-13 08:30:00
pYvotuLDIT
8881237
2018-02-13 10:32:00
KBuMzRpsko
user_client_log：保存了用户的客户端日志，每一条日志包含了用户ID、日志时间和日志内容。样例数据如下。
user_id
time
content
8881237
2018-02-13 00:30:00
click MpkvilgWSmhUuPn
8881237
2018-02-13 06:14:00
click OkTYNUHMqZzlDyL
8881237
2018-02-13 10:30:00
click OkTYNUHMqZzlDyL

对于每一条客户端日志，找出该用户在payment表里时间最接近的一条支付记录，将其中的支付内容和日志内容合并输出。达到如下效果。

user_id	time	content
8881237	2018-02-13 00:30:00	click MpkvilgWSmhUuPn, pay pYvotuLDIT
8881237	2018-02-13 06:14:00	click OkTYNUHMqZzlDyL, pay pYvotuLDIT
8881237	2018-02-13 10:30:00	click OkTYNUHMqZzlDyL, pay KBuMzRpsko

面对此类需求通常有如下2种解决方案：

使用内置Join。SQL伪代码如下。
```
SELECT
  p.user_id,
  p.time,
  merge(p.pay_info, u.content)
FROM
  payment p RIGHT OUTER JOIN user_client_log u
ON p.user_id = u.user_id and abs(p.time - u.time) = min(abs(p.time - u.time))
```
关联时需要知道相同user_id下的p.time与u.time差异最小的值，且聚合函数不能出现在关联条件上。因此，这个看似简单的需求，无法通过标准的关联操作实现。

使用UDJ方法实现。

注册UDJ函数。

配置新版本的SDK。

<dependency>
  <groupId>com.aliyun.odps</groupId>
  <artifactId>odps-sdk-udf</artifactId>
  <version>0.29.10-public</version>
  <scope>provided</scope>
</dependency>

编写UDJ代码，并将代码打包为odps-udj-example.jar。

package com.aliyun.odps.udf.example.udj;
import com.aliyun.odps.Column;
import com.aliyun.odps.OdpsType;
import com.aliyun.odps.Yieldable;
import com.aliyun.odps.data.ArrayRecord;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.udf.DataAttributes;
import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.UDJ;
import com.aliyun.odps.udf.annotation.Resolve;
import java.util.ArrayList;
import java.util.Iterator;
/** 对于右表的每个记录，找到最近的左表记录
 * 合并两条记录。
 */
@Resolve("->string,bigint,string")
public class PayUserLogMergeJoin extends UDJ {
  private Record outputRecord;
  /** 将在数据处理阶段之前调用。 用户可以实施这个方法做初始化工作。
   */
  @Override
  public void setup(ExecutionContext executionContext, DataAttributes dataAttributes) {
    //
    outputRecord = new ArrayRecord(new Column[]{
      new Column("user_id", OdpsType.STRING),
      new Column("time", OdpsType.BIGINT),
      new Column("content", OdpsType.STRING)
    });
  }
  /** 重写此方法以实现连接逻辑。
   * @param key 当前连接键。
   * @param 左表对应当前键的记录组。
   * @param right对应当前键的右表记录组。
   * @param output用于输出UDJ的结果。
   */
  @Override
  public void join(Record key, Iterator<Record> left, Iterator<Record> right, Yieldable<Record> output) {
    outputRecord.setString(0, key.getString(0));
    if (!right.hasNext()) {
      // 空右组，什么都不做。
      return;
    } else if (!left.hasNext()) {
      // 空左组。 输出右侧组的所有记录而不合并。
      while (right.hasNext()) {
        Record logRecord = right.next();
        outputRecord.setBigint(1, logRecord.getDatetime(0).getTime());
        outputRecord.setString(2, logRecord.getString(1));
        output.yield(outputRecord);
      }
      return;
    }
    ArrayList<Record> pays = new ArrayList<>();
    // 左侧记录组将从开始到结束迭代。
    // 对于右组的每个记录，迭代器无法重置。
    // 所以我们将左边的每个记录保存到ArrayList。
    left.forEachRemaining(pay -> pays.add(pay.clone()));
    while (right.hasNext()) {
      Record log = right.next();
      long logTime = log.getDatetime(0).getTime();
      long minDelta = Long.MAX_VALUE;
      Record nearestPay = null;
      // 迭代左边的所有记录，找到有的记录时间上的微小差异。
      for (Record pay: pays) {
        long delta = Math.abs(logTime - pay.getDatetime(0).getTime());
        if (delta < minDelta) {
          minDelta = delta;
          nearestPay = pay;
        }
      }
      // 将日志记录与最近的支付记录合并，并输出到结果。
      outputRecord.setBigint(1, log.getDatetime(0).getTime());
      outputRecord.setString(2, mergeLog(nearestPay.getString(1), log.getString(1)));
      output.yield(outputRecord);
    }
  }
  String mergeLog(String payInfo, String logContent) {
    return logContent + ", pay " + payInfo;
  }
  @Override
  public void close() {
  }
}

在MaxCompute中添加Jar包资源。
```
add jar odps-udj-example.jar;
```

在MaxCompute中注册UDF函数pay_user_log_merge_join。

create function pay_user_log_merge_join
  as 'com.aliyun.odps.udf.example.udj.PayUserLogMergeJoin'
  using 'odps-udj-example.jar';

准备示例数据。

创建示例表payment和user_client_log。

create table payment(user_id string,time datetime,pay_info string);
create table user_client_log(user_id string,time datetime,content string);

为示例表中插入数据。

--向payment表中插入数据。
INSERT OVERWRITE TABLE payment VALUES
('1335656', datetime '2018-02-13 19:54:00', 'PEqMSHyktn'),
('2656199', datetime '2018-02-13 12:21:00', 'pYvotuLDIT'),
('2656199', datetime '2018-02-13 20:50:00', 'PEqMSHyktn'),
('2656199', datetime '2018-02-13 22:30:00', 'gZhvdySOQb'),
('8881237', datetime '2018-02-13 08:30:00', 'pYvotuLDIT'),
('8881237', datetime '2018-02-13 10:32:00', 'KBuMzRpsko'),
('9890100', datetime '2018-02-13 16:01:00', 'gZhvdySOQb'),
('9890100', datetime '2018-02-13 16:26:00', 'MxONdLckwa')
;
--向user_client_log表中插入数据。
INSERT OVERWRITE TABLE user_client_log VALUES
('1000235', datetime '2018-02-13 00:25:36', 'click FNOXAibRjkIaQPB'),
('1000235', datetime '2018-02-13 22:30:00', 'click GczrYaxvkiPultZ'),
('1335656', datetime '2018-02-13 18:30:00', 'click MxONdLckpAFUHRS'),
('1335656', datetime '2018-02-13 19:54:00', 'click mKRPGOciFDyzTgM'),
('2656199', datetime '2018-02-13 08:30:00', 'click CZwafHsbJOPNitL'),
('2656199', datetime '2018-02-13 09:14:00', 'click nYHJqIpjevkKToy'),
('2656199', datetime '2018-02-13 21:05:00', 'click gbAfPCwrGXvEjpI'),
('2656199', datetime '2018-02-13 21:08:00', 'click dhpZyWMuGjBOTJP'),
('2656199', datetime '2018-02-13 22:29:00', 'click bAsxnUdDhvfqaBr'),
('2656199', datetime '2018-02-13 22:30:00', 'click XIhZdLaOocQRmrY'),
('4356142', datetime '2018-02-13 18:30:00', 'click DYqShmGbIoWKier'),
('4356142', datetime '2018-02-13 19:54:00', 'click DYqShmGbIoWKier'),
('8881237', datetime '2018-02-13 00:30:00', 'click MpkvilgWSmhUuPn'),
('8881237', datetime '2018-02-13 06:14:00', 'click OkTYNUHMqZzlDyL'),
('8881237', datetime '2018-02-13 10:30:00', 'click OkTYNUHMqZzlDyL'),
('9890100', datetime '2018-02-13 16:01:00', 'click vOTQfBFjcgXisYU'),
('9890100', datetime '2018-02-13 16:20:00', 'click WxaLgOCcVEvhiFJ')
;

在SQL中使用UDJ函数。

SELECT r.user_id, from_unixtime(time/1000) as time, content FROM (
SELECT user_id, time as time, pay_info FROM payment
) p JOIN (
SELECT user_id, time as time, content FROM user_client_log
) u
ON p.user_id = u.user_id
USING pay_user_log_merge_join(p.time, p.pay_info, u.time, u.content)
r
AS (user_id, time, content);

USING子句中的参数说明：

pay_user_log_merge_join是注册的UDJ在SQL中的函数名。
(p.time, p.pay_info, u.time, u.content)是UDJ中用到的左右表的列。
r是UDJ结果的别名，方便其他地方引用UDJ的结果。
(user_id, time, content)是UDJ产生的结果的列名。

运行结果如下。

+---------+------------+---------+
| user_id | time       | content |
+---------+------------+---------+
| 1000235 | 2018-02-13 00:25:36 | click FNOXAibRjkIaQPB |
| 1000235 | 2018-02-13 22:30:00 | click GczrYaxvkiPultZ |
| 1335656 | 2018-02-13 18:30:00 | click MxONdLckpAFUHRS, pay PEqMSHyktn |
| 1335656 | 2018-02-13 19:54:00 | click mKRPGOciFDyzTgM, pay PEqMSHyktn |
| 2656199 | 2018-02-13 08:30:00 | click CZwafHsbJOPNitL, pay pYvotuLDIT |
| 2656199 | 2018-02-13 09:14:00 | click nYHJqIpjevkKToy, pay pYvotuLDIT |
| 2656199 | 2018-02-13 21:05:00 | click gbAfPCwrGXvEjpI, pay PEqMSHyktn |
| 2656199 | 2018-02-13 21:08:00 | click dhpZyWMuGjBOTJP, pay PEqMSHyktn |
| 2656199 | 2018-02-13 22:29:00 | click bAsxnUdDhvfqaBr, pay gZhvdySOQb |
| 2656199 | 2018-02-13 22:30:00 | click XIhZdLaOocQRmrY, pay gZhvdySOQb |
| 4356142 | 2018-02-13 18:30:00 | click DYqShmGbIoWKier |
| 4356142 | 2018-02-13 19:54:00 | click DYqShmGbIoWKier |
| 8881237 | 2018-02-13 00:30:00 | click MpkvilgWSmhUuPn, pay pYvotuLDIT |
| 8881237 | 2018-02-13 06:14:00 | click OkTYNUHMqZzlDyL, pay pYvotuLDIT |
| 8881237 | 2018-02-13 10:30:00 | click OkTYNUHMqZzlDyL, pay KBuMzRpsko |
| 9890100 | 2018-02-13 16:01:00 | click vOTQfBFjcgXisYU, pay gZhvdySOQb |
| 9890100 | 2018-02-13 16:20:00 | click WxaLgOCcVEvhiFJ, pay MxONdLckwa |
+---------+------------+---------+

UDJ预排序功能

为了找到payment值相差最小的一条记录，需要反复对payment表的数据进行iterator遍历。所以事先将相同user_id的payment记录全部加载到了ArrayList，该方法适用于同一个用户一天之内的支付行为较少的场景。在其他场景中，有时同组内的数据量可能非常大，以至于无法在内存中存放，此时，您可以通过SORT BY预排序解决。

当某个用户的支付数据量非常大，导致无法将payment放在内存中时，如果组内所有数据已经按照时间排序，则只需要比较两边iterator最顶部的数据，即可实现该功能。

该方式主要是使用SORT BY子句对UDJ的数据进行预排序。在这个过程中，最多只需要同时缓存3条记录，就可以实现和之前算法相同的功能。Java UDJ代码如下。

@Override
public void join(Record key, Iterator<Record> left, Iterator<Record> right, Yieldable<Record> output) {
  outputRecord.setString(0, key.getString(0));
  if (!right.hasNext()) {
    return;
  } else if (!left.hasNext()) {
    while (right.hasNext()) {
      Record logRecord = right.next();
      outputRecord.setBigint(1, logRecord.getDatetime(0).getTime());
      outputRecord.setString(2, logRecord.getString(1));
      output.yield(outputRecord);
    }
    return;
  }
  long prevDelta = Long.MAX_VALUE;
  Record logRecord = right.next();
  Record payRecord = left.next();
  Record lastPayRecord = payRecord.clone();
  while (true) {
    long delta = logRecord.getDatetime(0).getTime() - payRecord.getDatetime(0).getTime();
    if (left.hasNext() && delta > 0) {
      //两个记录之间的时间差正在减少，我们仍然可以操作。
      //探索左侧组以尝试获得更小的增量。
      lastPayRecord = payRecord.clone();
      prevDelta = delta;
      payRecord = left.next();
    } else {
     //到达最小delta点。 检查最后的记录，
     //输出合并结果并准备处理下一条记录。
     //右组。
      Record nearestPay = Math.abs(delta) < prevDelta ? payRecord : lastPayRecord;
      outputRecord.setBigint(1, logRecord.getDatetime(0).getTime());
      String mergedString = mergeLog(nearestPay.getString(1), logRecord.getString(1));
      outputRecord.setString(2, mergedString);
      output.yield(outputRecord);
      if (right.hasNext()) {
        logRecord = right.next();
        prevDelta = Math.abs(
          logRecord.getDatetime(0).getTime() - lastPayRecord.getDatetime(0).getTime()
        );
      } else {
        break;
      }
    }
  }
}

user_id	time	pay_info
2656199	2018-02-13 22:30:00	gZhvdySOQb
8881237	2018-02-13 08:30:00	pYvotuLDIT
8881237	2018-02-13 10:32:00	KBuMzRpsko