AnalyticDB：UDF

Saat fungsi REMOTE_CALL() dipanggil, AnalyticDB for MySQL mengirimkan data ke Function Compute dalam format JSON. Setelah data diproses, Function Compute mengembalikan hasil dalam format JSON. Anda harus menulis fungsi yang ditentukan pengguna di Function Compute sesuai dengan format pengiriman dan pengembalian data yang ditentukan.

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

Pada contoh ini, fungsi yang ditentukan pengguna yang dibuat di Function Compute bernama ConcactNumberWithCompress. Berikut adalah kode contohnya:

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);
                    // Implementasikan logika bisnis Function Compute di sini.
                    // Anggaplah fungsi memiliki dua parameter input, a dan b. Setiap baris data yang dikirim dari AnalyticDB for MySQL berisi dua angka. Angka pertama adalah a, dan angka kedua adalah b.
                    if (row.size() == 2) {
                        result.add(testFunc(row.getInteger(0), row.getInteger(1)));
                    }
                    else {
                        throw new RuntimeException("row size is not 2");
                    }
                    // Logika bisnis Function Compute berakhir di sini.
                }

                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)
    {
        // Fungsi uji menggabungkan parameter input a dan b serta angka 1 menggunakan tanda ampersand (&).
        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;
    }
}

Contoh pemanggilan fungsi REMOTE_CALL di AnalyticDB for MySQL

• Jika tidak ada parameter global yang dikonfigurasi, panggil fungsi yang ditentukan pengguna ConcactNumberWithCompress untuk menggabungkan 1 dan 2 dengan tanda ampersand (&):

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

  Hasil berikut dikembalikan:

  1&2&1

• Jika tidak ada parameter global yang dikonfigurasi, panggil fungsi yang ditentukan pengguna ConcactNumberWithCompress untuk menggabungkan 3 dan 4 dengan tanda ampersand (&):

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

  Hasil berikut dikembalikan:

  3&4&1

• Jika parameter global telah dikonfigurasi, panggil fungsi yang ditentukan pengguna ConcactNumberWithCompress untuk menggabungkan 5 dan 6 dengan tanda ampersand (&):

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

  Hasil berikut dikembalikan:

  5&6&1

• AnalyticDB for MySQL mengompresi data ke format GZIP dan mengirimkannya ke Function Compute. Namun, fungsi yang ditentukan pengguna yang dibuat di Function Compute tidak berisi kode dekompresi. Akibatnya, Function Compute tidak dapat mengurai data tersebut.

• AnalyticDB for MySQL mengirimkan data yang tidak terkompresi ke Function Compute. Setelah diproses, Function Compute mengompresi hasilnya menggunakan format GZIP dan mengembalikannya ke AnalyticDB for MySQL. Akibatnya, AnalyticDB for MySQL tidak dapat mengurai data tersebut.