AnalyticDB:,

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

この例では、ConcactNumberWithCompressという名前のカスタム関数がfunction Computeに作成されます。 サンプルコード：

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);
                    // The beginning of the Function Compute business logic. 
                    // For example, if the custom function contains two input parameters a and b, each row of data that is sent by AnalyticDB for MySQL contains two data items a and b. 
                    if (row.size() == 2) {
                        result.add(testFunc(row.getInteger(0), row.getInteger(1)));
                    }
                    else {
                        throw new RuntimeException("row size is not 2");
                    }
                    // The end of the Function Compute business logic. 
                }

                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)
    {
        // Return input parameters a and b and a value of 1 by combining them with ampersands (&). 
        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はFunction ComputeにGZIP形式でデータを送信しますが、function Computeで作成されたカスタム関数には解凍コードは含まれません。 その結果、Function Computeはデータの解析に失敗します。

• AnalyticDB for MySQLは非圧縮データをFunction Computeに送信しますが、Function Computeは圧縮結果をGZIP形式でAnalyticDB for MySQLに送信します。 その結果、AnalyticDB for MySQLはデータの解析に失敗します。