AnalyticDB：自訂函數UDF

AnalyticDB for MySQL調用REMOTE_CALL函數後，會將資料以JSON格式發送到Function Compute服務；經過計算後，Function Compute服務會將計算結果以JSON形式返回。您需要根據資料發送的格式和計算結果返回的格式，在Function Compute服務中編寫自訂函數。

{
  "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" : [  ] 
}

本文樣本Function Compute服務中建立的自訂函數名稱為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);
                    // 從這裡開始實現Function Compute真正的商務邏輯。
                    // 假設這裡的函數有兩個入參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");
                    }
                    // Function Compute邏輯結束。
                }

                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函數樣本

• 在未設定全域配置的情況下，調用Function Compute服務中建立的自訂函數ConcactNumberWithCompress，將1和2使用&串連：

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

  返回結果：

  1&2&1

• 在未設定全域配置的情況下，調用Function Compute服務中建立的自訂函數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

• 在設定全域配置的情況下，調用Function Compute服務中建立的自訂函數ConcactNumberWithCompress，將5和6使用&串連：

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

  返回結果：

  5&6&1

• AnalyticDB for MySQL將資料使用GZIP格式壓縮後傳輸至Function Compute服務，但在Function Compute服務建立的自訂函數中沒有編寫解壓縮代碼，導致Function Compute服務無法解析資料。

• AnalyticDB for MySQL未開啟壓縮，將資料直接傳輸至Function Compute服務，Function Compute服務經過計算後，將計算結果使用GZIP格式壓縮後返回給AnalyticDB for MySQL，導致AnalyticDB for MySQL無法解析資料。