云原生数据仓库AnalyticDB：自定义函数UDF

AnalyticDB for MySQL调用REMOTE_CALL函数后，会将数据以JSON格式发送到函数计算服务；经过计算后，函数计算服务会将计算结果以JSON形式返回。您需要根据数据发送的格式和计算结果返回的格式，在函数计算服务中编写自定义函数。

{
  "rowCount": 3, 
  "compressed":true, 
  "data": 
  [
      [1, "a", "2023-08-22 11:30:00"],
      [2, "b", "2023-08-22 12:30:00"],
      [3, "c", null]
  ]
}

{
  "success": true,   
  "message": "",  
  "result" : [  ] 
}

本文示例函数计算服务中创建的自定义函数名称为ConcactNumberWithCompress，示例代码如下：

public class App
        implements StreamRequestHandler, FunctionInitializer
{
    public static final Logger log = Logger.getLogger(App.class.getName());

    public void initialize(Context context)
            throws IOException
    {
        // TODO
    }

    @Override
    public void handleRequest(
            InputStream inputStream, OutputStream outputStream, Context context)
            throws IOException
    {
        InputStream notCompressedInputStream1;
        notCompressedInputStream1 = tryUnCompress(inputStream);

        JSONObject response = new JSONObject();
        try {
            JSONObject requestJson = JSONObject.parseObject(IOUtils.toString(notCompressedInputStream1));
            if (requestJson.containsKey("data")) {
                JSONArray result = new JSONArray();
                JSONArray data = requestJson.getJSONArray("data");
                for (int i = 0; i < data.size(); i++) {
                    JSONArray row = data.getJSONArray(i);
                    // 从这里开始实现函数计算真正的业务逻辑。
                    // 假设这里的函数有两个入参a和b，则AnalyticDB for MySQL发来的数据每行都会有两个数，第一个数是a，第二个数是b。
                    if (row.size() == 2) {
                        result.add(testFunc(row.getInteger(0), row.getInteger(1)));
                    }
                    else {
                        throw new RuntimeException("row size is not 2");
                    }
                    // 函数计算逻辑结束。
                }

                response.put("result", result);
                response.put("success", true);
                response.put("message", "");
            }
            else {
                response.put("success", false);
                response.put("message", "no data inside");
            }
        }
        catch (Exception e) {
            log.info("error happened" + e.getMessage());
            response.put("success", false);
            response.put("message", e.getMessage());
        }

        outputStream.write(tryCompress(response.toJSONString().getBytes()));
    }

    private String testFunc(int a, int b)
    {
        // 测试函数将入参a、b和1使用&符号连接。
        return String.valueOf(a) + '&' + b + '&' + 1;
    }

    public static byte[] tryCompress(byte[] bytes)
    {
        ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
        try {
            GZIPOutputStream gzipOutputStream = new GZIPOutputStream(byteOutputStream);
            gzipOutputStream.write(bytes);
            gzipOutputStream.close();
        }
        catch (IOException e) {
            throw new RuntimeException(e);
        }

        return byteOutputStream.toByteArray();
    }

    public static InputStream tryUnCompress(InputStream inputStream)
            throws IOException
    {
        GZIPInputStream gzipInputStream;
        gzipInputStream = new GZIPInputStream(inputStream);
        return gzipInputStream;
    }
}

AnalyticDB for MySQL调用REMOTE_CALL函数示例

• 在未设置全局配置的情况下，调用函数计算服务中创建的自定义函数ConcactNumberWithCompress，将1和2使用&连接：

  SELECT remote_call('varchar','ConcactNumberWithCompress','{endpoint:"1234567890000****.cn-zhangjiakou-internal.fc.aliyuncs.com"}',1,2);

  返回结果：

  1&2&1

• 在未设置全局配置的情况下，调用函数计算服务中创建的自定义函数ConcactNumberWithCompress，将3和4使用&连接：

  SELECT remote_call('varchar','ConcactNumberWithCompress','{endpoint:"1234567890000****.cn-zhangjiakou-internal.fc.aliyuncs.com",compressed:false,max_batch_size:5000000}',3,4);

  返回结果：

  3&4&1

• 在设置全局配置的情况下，调用函数计算服务中创建的自定义函数ConcactNumberWithCompress，将5和6使用&连接：

  SELECT remote_call('varchar','ConcactNumberWithCompress',null,5,6);

  返回结果：

  5&6&1

• AnalyticDB for MySQL将数据使用GZIP格式压缩后传输至函数计算服务，但在函数计算服务创建的自定义函数中没有编写解压缩代码，导致函数计算服务无法解析数据。

• AnalyticDB for MySQL未开启压缩，将数据直接传输至函数计算服务，函数计算服务经过计算后，将计算结果使用GZIP格式压缩后返回给AnalyticDB for MySQL，导致AnalyticDB for MySQL无法解析数据。