Platform For AI:SDK for Java

PredictClient class

PredictClient is the main client class. Use it to configure service connection parameters, send prediction requests, and handle responses.

client.setRetryConditions(
    EnumSet.of(
        RetryCondition.READ_TIMEOUT,
        RetryCondition.RESPONSE_5XX
    )
);

HttpConfig class

HttpConfig configures the underlying HTTP connection: timeouts, thread counts, and the connection pool.

TFRequest class

TFRequest builds input data for TensorFlow models saved in SavedModel format.

QueueClient class

QueueClient interacts with the EAS queue service to produce, consume, and manage data in asynchronous inference workflows.

// Set the group ID to the service name
QueueUser u = new QueueUser(UUID.randomUUID().toString(), "<service_name>");
QueueClient inputQueue = new QueueClient(queueEndpoint, inputQueueName, queueToken, new HttpConfig(), u);

// Query the queue position of a specific record
System.out.println(inputQueue.search(index));

DataFrame class

DataFrame wraps a single data record from the queue service.

Send a string request

Use this pattern when your model service accepts and returns plain strings, for example with PMML models deployed using a custom processor.

import com.aliyun.openservices.eas.predict.http.PredictClient;
import com.aliyun.openservices.eas.predict.http.HttpConfig;

public class TestString {
    public static void main(String[] args) throws Exception {
        // Initialize the client. One PredictClient instance handles all requests to a service.
        // Do not create a new instance per request.
        PredictClient client = new PredictClient(new HttpConfig());
        client.setToken("YWFlMDYyZDNmNTc3M2I3MzMwYmY0MmYwM2Y2MTYxMTY4NzBkNzdj****");
        // For VPC direct connection, use setDirectEndpoint instead:
        // client.setDirectEndpoint("pai-eas-vpc.cn-shanghai.aliyuncs.com");
        client.setEndpoint("182848887922****.vpc.cn-shanghai.pai-eas.aliyuncs.com");
        client.setModelName("scorecard_pmml_example");

        String request = "[{\"money_credit\": 3000000}, {\"money_credit\": 10000}]";
        System.out.println(request);

        try {
            String response = client.predict(request);
            System.out.println(response);
        } catch (Exception e) {
            System.err.printf("Prediction failed. Error code: %d, message: %s%n",
                client.getErrorCode(), client.getErrorMessage());
            e.printStackTrace();
        } finally {
            client.shutdown();
        }
    }
}

Send a TensorFlow request

For TensorFlow models, use TFRequest to build the input and TFResponse to read the output.

import java.util.List;

import com.aliyun.openservices.eas.predict.http.PredictClient;
import com.aliyun.openservices.eas.predict.http.HttpConfig;
import com.aliyun.openservices.eas.predict.request.TFDataType;
import com.aliyun.openservices.eas.predict.request.TFRequest;
import com.aliyun.openservices.eas.predict.response.TFResponse;

public class TestTF {
    // Build a TFRequest for the "predict_images" signature.
    // Input: 784 float values (1x784 tensor), all initialized to 0.0.
    // Output: the "scores" tensor.
    public static TFRequest buildPredictRequest() {
        TFRequest request = new TFRequest();
        request.setSignatureName("predict_images");
        float[] content = new float[784];
        request.addFeed("images", TFDataType.DT_FLOAT, new long[]{1, 784}, content);
        request.addFetch("scores");
        return request;
    }

    public static void main(String[] args) throws Exception {
        PredictClient client = new PredictClient(new HttpConfig());
        // For VPC direct connection, use setDirectEndpoint instead:
        // client.setDirectEndpoint("pai-eas-vpc.cn-shanghai.aliyuncs.com");
        client.setEndpoint("182848887922****.vpc.cn-shanghai.pai-eas.aliyuncs.com");
        client.setModelName("mnist_saved_model_example");
        client.setToken("YTg2ZjE0ZjM4ZmE3OTc0NzYxZDMyNmYzMTJjZTQ1YmU0N2FjMTAy****");

        int count = 1000;
        long startTime = System.currentTimeMillis();

        for (int i = 0; i < count; i++) {
            try {
                TFResponse response = client.predict(buildPredictRequest());
                List<Float> result = response.getFloatVals("scores");
                System.out.print("Predict result: [");
                for (int j = 0; j < result.size(); j++) {
                    System.out.print(result.get(j).floatValue());
                    if (j != result.size() - 1) {
                        System.out.print(", ");
                    }
                }
                System.out.print("]\n");
            } catch (Exception e) {
                System.err.printf("Prediction failed. Error code: %d, message: %s%n",
                    client.getErrorCode(), client.getErrorMessage());
                e.printStackTrace();
            }
        }

        long endTime = System.currentTimeMillis();
        System.out.println("Total time: " + (endTime - startTime) + " ms");
        client.shutdown();
    }
}

Use the queue service

The queue service manages asynchronous inference through an input queue and an output queue. Your application writes requests to the input queue; the inference service reads them, processes them, and writes results to the output queue.

import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.eas.predict.http.HttpConfig;
import com.aliyun.openservices.eas.predict.http.QueueClient;
import com.aliyun.openservices.eas.predict.queue_client.QueueUser;
import com.aliyun.openservices.eas.predict.queue_client.WebSocketWatcher;

public class DemoWatch {
    public static void main(String[] args) throws Exception {
        String queueEndpoint = "18*******.cn-hangzhou.pai-eas.aliyuncs.com";
        String inputQueueName = "test_queue_service";
        String sinkQueueName = "test_queue_service/sink";
        String queueToken = "test-token";

        // Create clients for the input and output queues.
        // The inference service reads from inputQueue and writes results to sinkQueue.
        QueueClient inputQueue =
            new QueueClient(queueEndpoint, inputQueueName, queueToken, new HttpConfig(), new QueueUser());
        QueueClient sinkQueue =
            new QueueClient(queueEndpoint, sinkQueueName, queueToken, new HttpConfig(), new QueueUser());

        // Clear any existing data. Use with caution.
        inputQueue.clear();
        sinkQueue.clear();

        // Write 10 records to the input queue.
        // To set a priority, use: inputQueue.put(data.getBytes(), 0L, null)
        // Priority 0 = low (default), 1 = high.
        int count = 10;
        for (int i = 0; i < count; ++i) {
            String data = Integer.toString(i);
            inputQueue.put(data.getBytes(), null);
        }

        // Subscribe to the output queue with a sending window of 5.
        // The queue service pauses sending when uncommitted records exceed the window size.
        // Note: when autoCommit is true, the window parameter is invalid.
        // To configure retries: new WatchConfig(retryCount, retryIntervalSeconds)
        // To retry indefinitely: new WatchConfig(true, retryIntervalSeconds)
        WebSocketWatcher watcher = sinkQueue.watch(0L, 5L, false, true, null);

        // Read results. getDataFrame() blocks until a record is available.
        for (int i = 0; i < count; ++i) {
            try {
                byte[] data = watcher.getDataFrame().getData();
                System.out.println("[watch] data = " + new String(data));
            } catch (RuntimeException ex) {
                System.out.println("[watch] error = " + ex.getMessage());
                break;
            }
        }

        // Close the watcher before creating another one.
        // Each client supports only one active watcher at a time.
        watcher.close();

        Thread.sleep(2000);
        JSONObject attrs = sinkQueue.attributes();
        System.out.println(attrs.toString());

        inputQueue.shutdown();
        sinkQueue.shutdown();
    }
}

Compress request data

For large payloads, compress requests before sending them to reduce network overhead. The SDK supports Zlib and Gzip compression.

First, enable decompression on the server side by setting rpc.decompressor in the service deployment configuration:

{
  "metadata": {
    "rpc": {
      "decompressor": "zlib"
    }
  }
}

Then set the matching compressor on the client:

package com.aliyun.openservices.eas.predict;

import com.aliyun.openservices.eas.predict.http.Compressor;
import com.aliyun.openservices.eas.predict.http.PredictClient;
import com.aliyun.openservices.eas.predict.http.HttpConfig;

public class TestString {
    public static void main(String[] args) throws Exception {
        PredictClient client = new PredictClient(new HttpConfig());
        client.setEndpoint("18*******.cn-hangzhou.pai-eas.aliyuncs.com");
        client.setModelName("echo_compress");
        client.setToken("YzZjZjQwN2E4NGRkMDMxNDk5NzhhZDcwZDBjOTZjOGYwZDYxZGM2****");
        // Set Compressor.Gzip to use Gzip instead.
        client.setCompressor(Compressor.Zlib);

        String request = "[{\"money_credit\": 3000000}, {\"money_credit\": 10000}]";
        System.out.println(request);

        String response = client.predict(request);
        System.out.println(response);

        client.shutdown();
    }
}