AnalyticDB:UDFs

REMOTE_CALL() 関数が呼び出されると、AnalyticDB for MySQL はデータを 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.getJSONJSONArray(i);
                    // ここに Function Compute のビジネスロジックを実装します。
                    // この関数には a と b の 2 つの入力パラメーターがあると仮定します。AnalyticDB for MySQL から送信される各データ行には 2 つの数値が含まれます。最初の数値が a で、2 番目の数値が 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;
    }
}

REMOTE_CALL 関数を AnalyticDB for MySQL で呼び出す例

• グローバルパラメーターが設定されていない場合、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 フォーマットに圧縮して Function Compute に送信しますが、Function Compute で作成されたユーザー定義関数に展開コードが含まれていないため、Function Compute はデータを解析できません。

• AnalyticDB for MySQL は非圧縮データを Function Compute に送信します。処理後、Function Compute は結果を GZIP フォーマットを使用して圧縮し、AnalyticDB for MySQL に返します。その結果、AnalyticDB for MySQL はデータを解析できません。