This document describes how to use the API interface to connect to MatrixGate to achieve high-speed import of data.
MatrixGate provides an HTTP API to the outside world, supporting various programming languages to import data into the YMatrix database through the HTTP interface. The MatrixGate HTTP protocol format and response code are as follows:
MatrixGate HTTP protocol format
| Protocol Type | Protocol Format | Usage and Examples |
| --- | --- |
| URL | http://\
Timestamp|ID]|C1|C2|..|Cn | Body format The first behavior of data loading target table, SchemeName can be omitted, default is Public, TableName is a required item, and the second row starts with a time-series data row. Each row corresponds to a row of the target table. The | separator is used between columns, and the \n separator is used between rows. The first field of each row is a timestamp, the format is UNIX timestamp is accurate to seconds, see the description of --time-format. The second field of each row is TagID, integer. The third field to the last field of each row is the column corresponding to the target table. It is recommended that the DDL definition of the target table also follows the column order of (Timestamp, TagID, C1, C2,…, Cn) |
MatrixGate HTTP response code
| Response Code | Response Code Meaning | Remarks |
| --- | --- |
| 200 | StatusOK | Some data format is wrong, and the response Body will contain the wrong line with error message, such as: At line: 2missing data for column "c3" |
| 204 | StatusNoContent | Data loaded successfully to MatrixGate |
| 400 | StatusBadRequest | Data request errors, such as POST BODY format error, target table does not exist, data compression format does not match the HTTP request header, etc. |
| 405 | StatusMethodNotAllowed | HTTP Non-POST Request |
| 408 | StatusTimeout | Request timeout |
| 500 | StatusIntervalServerError | Database error, data loading failed, response Body contains detailed error information |
| 503 | StatusServiceUnavailable | MatrixGate rejects requests, such as exceeding the maximum number of connections, or MatrixGate is closing, etc |
testtable in the demo database.=# CREATE TABLE testtable (
time TIMESTAMP WITH TIME ZONE,
tagid INT,
c1 INT,
c2 INT,
c3 INT
)USING MARS3
DISTRIBUTED BY (tagid)
ORDER BY (time,tagid);mxgate.conf configuration file.$ mxgate config --db-database testdb \
--db-master-host localhost \
--db-master-port 5432 \
--db-user mxadmin \
--db-password 123123 \
--target public.testtable \
--format csv \
--time-format unix-second \
--delimiter '|' \
--parallel 256 \
--stream-prepared 3 \
--interval 250 \
--transform plain \
> mxgate.confdata.txt:$ vi data.txt
public.testtable
1603777821|1|101|201|301
1603777822|2|102|202|302
1603777823|3|103|203|303mxgate.conf.$ mxgate --config mxgate.conf$ curl http://localhost:8086/ -X POST -H 'Content-Type: text/plain' --data-binary "@data.txt"demo=# SELECT extract(epoch FROM "time"), * FROM testtable;
date_part | time | tagid | c1 | c2 | c3
----------------------------------------------------------------------------------------------------------------------------------
1603777821 | 2020-10-27 13:50:21+08 | 1 | 101 | 201 | 301
1603777822 | 2020-10-27 13:50:22+08 | 2 | 102 | 202 | 302
1603777823 | 2020-10-27 13:50:23+08 | 3 | 103 | 203 | 303
(3 rows)SDK (Software Development Kit) refers to a software development toolkit, which can free developers from non-business logic functional development, greatly improving development efficiency and experience, and increasing ease of use.
You can introduce the SDK JAR package in the following ways: (1). Get it directly from Maven remote repository (2). Get it directly from Gradle remote repository (3). Manually download the JAR package to the local area and then manually import it
Note! Choose any of the above methods**. It is recommended to use (1) or (2) to directly introduce SDK from Maven or Gradle remote warehouses, which is efficient and convenient. (3) Please refer to MatrixGate FAQ: 21. It is recommended to search the keyword
JAVA SDKdirectly, because it is not commonly used, I will not explain it in detail here.
(1). Call Maven remote repository to automatically download SDK packages Configure the following dependencies in the pom.xml file of your JAVA project.
<dependencies>
<dependency>
<groupId>cn.ymatrix</groupId>
<artifactId>mxgate-sdk-java</artifactId>
<version>1.1.2</version>
</dependency>
</dependencies>(2). Use Gradle Remote Repository to Introduction SDK Dependencies
repositories {
mavenCentral()
}
dependencies {
implementation 'cn.ymatrix:mxgate-sdk-java:1.0.20'
}YMatrix supports writing data through gRPC and HTTP. The specific description is as follows:
Note! You need to use the YMatrix version that supports gRPC, i.e. 4.6.1 and later.
Write the mxgate configuration file mxgate_grpc.conf on the Master and start mxgate.
Note! Before using mxgate, you need to create tables in the database first. In the example, the Master host is mdw, the database is
demo, and the data table istest_table_transfer.
The code is as follows:
# Create mxgate config file
[mxadmin@mdw ~]$ mxgate config \
--source grpc \
--db-database demo \
--target public.test_table_transfer \
--time-format raw \
--grpc-port 8087 \
--format csv \
> mxgate_grpc.conf
# Start mxgate
[mxadmin@mdw ~]$ mxgate start --config mxgate_grpc.confAs shown in the example code, to use the SDK to write data to mxgate, you need to specify that the source parameter of mxgate is grpc and format is csv. At the same time, since the SDK needs to know where to write the data, it is necessary to specify the grpc-port port number in the configuration file, 8087 in the example.
Note! Starting mxgate through HTTP still requires specifying the gRPC port number of mxgate, because the SDK still obtains database table meta information from mxgate in gRPC, but the way to write data is switched to HTTP.
Write the mxgate configuration file mxgate_http.conf on the Master and start mxgate. In the example, the Master host name is mdw, and the code is as follows:
# Create the mxgate config file.
[mxadmin@mdw ~]$ mxgate config \
--source http \
--db-database demo \
--target public.test_table_transfer \
--time-format raw \
--grpc-port 8087 \
--format csv \
> mxgate_http.conf
# Start mxgate
[mxadmin@mdw ~]$ mxgate start --config mxgate_http.confIn this step, two types of transmission methods will be introduced, asynchronous and synchronous, and you can choose according to your needs.
The example table structure is as follows:
Partitioned table "public.test_table_transfer"
Column | Type | Collation | Nullable | Default
--------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
ts | timestamp without time zone | | |
tag | integer | | not null |
c1 | double precision | | |
c2 | double precision | | |
c3 | double precision | | |
c4 | double precision | | |
c5 | text | | |
c6 | text | | |
c7 | text | | |
c8 | text | | |
Partition key: RANGE (ts)First, perform global initialization settings:
// Set the log level, default INFO.
MxLogger.loggerLevel(LoggerLevel.INFO);
// The log will be output to stdout by default. If you do not want to output, you can pass false in the following API.
MxLogger.enableStdout(true);
// The default log file path and naming format of SDK is /tmp/mxgate_sdk_java_2022-08-26_133403.log,
// Users can customize the file path and file name of the output log file.
MxLogger.writeToFile("/tmp/mxgate_sdk_java.log");Asynchronous sending the Tuple append to a queue inside the SDK, and then sends data via an asynchronous HTTP request. First, you need to start MxBuilder, which is singleton mode and is globally unique.
Note! Only build once in a project. The relevant configuration can be performed before initializing the MxBuilder.
MxBuilder builder = MxBuilder.newBuilder() .withDropAll(false) // If used for testing, set to true to drop data instead of sending to mxgate; set to false to send data to mxgate .withCacheCapacity(100000) // Size of the queue for temporarily storing tuples in batches .withCacheEnqueueTimeout(2000) // Timeout for tuples to be enqueued if the queue is full. Throws IllegalStateException if exceeded .withConcurrency(10) // Number of threads writing data to mxgate concurrently .withRequestTimeoutMillis(3000) // Timeout for each data write request (in milliseconds) .withMaxRetryAttempts(3) // Number of retries for each write request if an issue occurs .withRetryWaitDurationMillis(3000) // Interval between retries (currently fixed) .withRequestType(RequestType.WithHTTP) // SDK supports sending data to mxgate via HTTP and gRPC. Corresponding configurations are: RequestType.WithHTTP, RequestType.WithGRPC .withCircuitBreaker() // Use the built - in circuit breaker. If the failure rate or slow request rate reaches the threshold, it will be activated. After activation, data sending to mxgate will be paused for 30 seconds, and tuples cannot be appended .withMinimumNumberOfCalls(1) // Minimum number of calls for the circuit breaker to take effect (must be >= 1, default is 10) .withSlidingWindowSize(10) // Size of the sliding window for calculating the failure rate (must be >= 1, default is 100) .withFailureRateThreshold(60.0f) // Failure rate threshold (must be >0 and <= 100). If the failure rate reaches this threshold, the circuit breaker will be activated .withSlowCallDurationThresholdMillis(1000) // Threshold for slow request duration (milliseconds). Requests exceeding this duration are considered slow (note that this duration should be less than the request timeout) .withSlowCallRateThreshold(80.0f) // Slow request rate threshold. If the slow request rate reaches this threshold, the circuit breaker will be activated // .withRequestAsync(true) // Uncomment this line to enable asynchronous mode for sending data to mxgate (optional) // .withConcurrency(20) // Typically, asynchronous mode only requires tens of concurrency to achieve the same or even higher throughput as synchronous mode (optional) // MxBuilder's builder pattern has added the following API to adjust the concurrency of CSV parallel transformation (at the MxClient Group level) // .withCSVConstructionParallel(100) // (Supported from v1.1.0, uncomment this line to use) .build(); // MxBuilder has added a singleton API. After MxBuilder is successfully built, you can obtain the globally unique singleton instance of MxBuilder at any location using the following API (supported from v1.1.2) // MxBuilder.instance(); // (Supported from v1.1.2, uncomment this line to use) // builder.getTupleCacheSize(); // Uncomment this line to get the number of remaining Tuples in the SDK's internal Tuple cache in real - time (optional)The connect method of Builder receives four parameters: (1). The mxgate process provides the hostname (IP address) and port number where the specific service is located, which is used to receive data and database table meta information. For example, for the HTTP method, it is
http://localhost:8086/; for the gRPC method, it islocalhost:8087(2). schema. (3). table name. (4). callback. If the builder can successfully connect to mxgate, the onSuccess method will callback and return an MxClient instance for data writing. If the connection fails, onFailure will callback, and failureMsg will explain the reason for the failure.
The specific code is as follows. Remember to replace the port number, pattern name, and table name with the actual content.
// Asynchronous method (this feature is supported starting from v1.0.13)
builder.connect("http://localhost:8086/", "localhost:8087", "public", "test_table", new ConnectionListener() {
@Override
public void onSuccess(MxClient client) {
sendData(client);
}
@Override
public void onFailure(String failureMsg) {
}
});Through the onSuccess() callback function of ConnectionListener, you can get the MxClient instance, and then use the API provided by MxClient to implement the function of writing data to mxgate.
// Synchronization method (this feature is supported starting from v1.0.13)
MxClient client = builder.connect(httpHost, gRPCHost, schema, table);/*
* v1.1.0 improves the scalability of MxClient, and MxBuilder provides several new APIs. The MxClient obtained through this API belongs to a MxClient Group.
* You can freely define the concurrency parameters of MxClient Group Number, Tuples cache capability, and TuplesConsumer through these APIs.
* (This feature is supported starting from v1.1.0)
*/
// MxClient client = builder.connectWithGroup(dataSendingHost, metadataHost, schema, table, 10);
// MxClient client = builder.connectWithGroup(dataSendingHost, metadataHost, schema, table, 1, 1000, 3000, 10);// (This function starts with v1.1.0, if you use to remove the comment content of this line)
/*
* For APIs starting with skip, the process of connecting to back-end services to obtain database table meta information will be skipped when calling.
* This type of MxClient can only obtain lightweight Tuples through generateEmptyTupleLite().
* (This feature is supported starting from v1.1.0)
*/
// MxClient client = mxBuilder.skipConnectWithGroup(dataSendingHost, metadataHost, schema, table, delimiter, 1);
// MxClient client = mxBuilder.skipConnectWithGroup(dataSendingHost, metadataHost, schema, table, delimiter, 1, 1000, 3000, 1);Note! MxClient is thread-safe, but if used in concurrent multithreading, in order to ensure optimal performance, it is best to have a separate MxClient in each thread, that is, multiple MxClient instances are returned through MxBuilder multiple connects.
private void sendData(MxClient client) { if (client != null) { /* * MxClient will accumulate a batch of Tuples as a micro batch and send them to mxgate. * This API sets the waiting time accumulated for each microbatch, with a default of 2000 million seconds. * That is, each 2s will try to send a batch of data to mxgate if the 2s time period is * No data is written, no sending */ client.withIntervalToFlushMillis(5000);
/*
* Set how many Tuple bytes accumulated to be sent as a microbatch, even if the time has not been reached
* The set flush interval, the accumulated micro-batch bytes will also be sent if it reaches the number of micro-batch bytes.
* The time of flush interval has reached, and the number of bytes that are not accumulated enough to be set will also be sent
*/
client.withEnoughBytesToFlush(500);
/*
* Compared withEnoughBytesToFlush, when appendTuple
* Performance is improved because it avoids calculating the number of bytes.
* Depending on the specific use scenario, if withEnoughBytesToFlush
* It can also meet the performance requirements, so each flush, the data volume will be more uniform.
* withEnoughLinesToFlush will have higher priority than withEnoughBytesToFlush
*/
client.withEnoughLinesToFlush(10000);
/*
* MxClient has added the following API to adjust the number of Tuples for each CSV conversion subtask.
* For example, there are 10,000 Tuples in Flush at a time. Set BatchSize = 2000 through the following API.
* Then the CSV conversion of these 10,000 Tuples will be divided into 5 subtasks, each subtask handles 2,000 Tuples and executes concurrently.
* (This feature is supported starting from v1.1.0)
*/
client.withCSVConstructionBatchSize(2000);
/*
* Each MxClient has a private object pool.
* When using it, you need to set the size of the object pool according to the number of Tuples in each MxClient flush.
*/
// client.useTuplesPool(poolSize);
/*
* MxClient supports compression and needs to be used with mxgate v4.7.6 and higher
*/
// client.withCompress();
/*
* For HTTP requests, you can not use base64 encoding, and gRPC needs to do base64 encoding
*/
// client.withBase64Encode4Compress();
/*
* MxClient can register a DataPostListener, and the success and failure of each batch of data are sent.
* They will all callback in onSuccess and onFailure. You can understand which Tuple writes succeed and which Tuple writes fail
*/
client.registerDataPostListener(new DataPostListener() {
@Override
public void onSuccess(Result result) {
System.out.println(CUSTOMER_LOG_TAG + "Send tuples success: " + result.getMsg());
System.out.println(CUSTOMER_LOG_TAG + "Succeed lines onSuccess callback " + result.getSucceedLines());
}
@Override
public void onFailure(Result result) {
/*
* result.getErrorTuplesMap() key-value pair containing the line of error and the cause of the error Tuple -> String
* Key is the wrong line, value is the wrong reason
*/
for (Map.Entry<Tuple, String> entry : result.getErrorTuplesMap().entrySet()) {
l.error(CUSTOMER_LOG_TAG + "error tuple of table={}, tuple={}, reason={}", entry.getKey().getTableName(), entry.getKey(), entry.getValue());
for (Column c : entry.getKey().getColumns()) {
l.error(CUSTOMER_LOG_TAG + "error entry columns {} = {}", c.getColumnName(), c.getValue());
}
}
System.out.println(result.getSuccessLines());
}});
Return an empty Tuple object for assignment via MxClinet's generateEmptyTuple API.Tuple tuple1 = client.generateEmptyTuple(); Tuple tuple2 = client.generateEmptyTuple();
Assign a value to an empty Tuple in the form of Key -> Value. Key is the column name corresponding to the database table; Value is the value corresponding to this field. If some field values are allowed to be null or have default values, you can not add the value of the field, and the SDK will automatically fill in the default or empty values.
The meta information of the database tables obtained through this API is no longer maintained within the Tuple. It is a lighter-weight Tuple. Therefore, when calling MxClient.appendTuple() to append to MxClient, it eliminates the work of checking the internal data legitimacy of Tuple based on the database table meta information, so that the Tuple data can be flushed as fast as possible.// Tuple tuple = client.generateEmptyTupleLite();// (This function starts with v1.1.0, if you use it to remove the comment content of this line)
>***Note!***
When performing the add column operation to this lightweight Tuple, you need to manually maintain the order of the `addColumn() key -> value` pairs to ensure that the order corresponds to the order of columns in the database table.
For example, the data table column order is as follows:Column | Type | Collation | Nullable | Default --------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- ts | timestamp without time zone | | | tag | integer | | not null | c1 | double precision | | | c2 | double precision | | | c3 | double precision | | | c4 | double precision | | | c5 | text | | | c6 | text | | | c7 | text | | | c8 | text | | |
Then the order of adding columns must be consistent with the order of columns in the database table. In this example, the following is: tuple1.addColumn("ts", "2022-05-18 16:30:06");
tuple1.addColumn("tag", 102020030);
tuple1.addColumn("c1", 1.1);
tuple1.addColumn("c2", 2.2);
tuple1.addColumn("c3", 3.3);
tuple1.addColumn("c4", 4.4);
tuple1.addColumn("c5", "中文字符测试-1");
tuple1.addColumn("c6", "lTxFCVLwcDTKbNbjau_c6");
tuple1.addColumn("c7", "lTxFCVLwcDTKbNbjau_c7");
tuple1.addColumn("c8", "lTxFCVLwcDTKbNbjau_c8");
tuple2.addColumn("ts", "2022-05-18 16:30:06");
tuple2.addColumn("tag", 102020030);
tuple2.addColumn("c1", 1.1);
tuple2.addColumn("c2", 2.2);
tuple2.addColumn("c3", 3.3);
tuple2.addColumn("c4", 4.4);
tuple2.addColumn("c5", "中文字符测试-2");
tuple2.addColumn("c6", "lTxFCVLwcDTKbNbjau_c26");
tuple2.addColumn("c7", "lTxFCVLwcDTKbNbjau_c27");
tuple2.addColumn("c8", "lTxFCVLwcDTKbNbjau_c28");Finally, add the assigned Tuple append to MxClient.client.appendTuples(tuple1, tuple2);
> ***Note!***
MxClient provides multiple APIs, which can Append one Tuple at a time, Append multiple Tuples at a time, or a Tuples List. For example: `client.appendTuple();`, `client.appendTupleList();`; you can also call the flush() method of MxClient to manually send data, no matter how many Tuple data is written at this time, such as `client.flush();`.
- Send data synchronously using MxClient
Synchronous sending means synchronous sending Tuple to mxgate for easier subsequent processing (such as submitting Kafka offset).Tuple tuple1 = client.generateEmptyTuple(); tuple1.addColumn("ts", "2022-05-18 16:30:06"); tuple1.addColumn("tag", 102020030); tuple1.addColumn("c1", 1.1); tuple1.addColumn("c2", 2.2); tuple1.addColumn("c3", 3.3); tuple1.addColumn("c4", 4.4); tuple1.addColumn("c5", "lTxFCVLwcDTKbNbjau_c5"); tuple1.addColumn("c6", "lTxFCVLwcDTKbNbjau_c6"); tuple1.addColumn("c7", "lTxFCVLwcDTKbNbjau_c7"); tuple1.addColumn("c8", "lTxFCVLwcDTKbNbjau_c8");
/*
#### 3.1.2 MatrixGate HTTP API Java Exampleimport java.io.BufferedReader; import java.io.DataOutputStream; import java.io.InputStreamReader; import java.net.HttpURLConnection; import java.net.URL;
public class MxgateExample { public static void main(String[] args) throws Exception { MxgateExample http = new MxgateExample(); http.sendingPostRequest(); }
/*
* HTTP Post request
*/
private void sendingPostRequest() throws Exception {
/*
* mxgate listens on port 8086 of localhost
*/
String url = "http://localhost:8086/";
URL obj = new URL(url);
HttpURLConnection con = (HttpURLConnection) obj.openConnection();
/*
* Setting basic post request
*/
con.setRequestMethod("POST");
con.setRequestProperty("Content-Type","text/plain");
String postJsonData = "public.testtable\n1603777821|1|101|201|301\n1603777822|2|102|202|302\n1603777823|3|103|203|303";
con.setDoOutput(true);
DataOutputStream wr = new DataOutputStream(con.getOutputStream());
/*
* When the data is in Chinese, it can be encoded by postJsonData.getBytes("UTF-8")
*/
wr.write(postJsonData.toString().getBytes("UTF-8"));
wr.flush();
wr.close();
int responseCode = con.getResponseCode();
System.out.println("Sending 'POST' request to URL : " + url);
System.out.println("Post Data : " + postJsonData);
System.out.println("Response Code : " + responseCode);
BufferedReader in = new BufferedReader(new InputStreamReader(con.getInputStream()));
String output;
StringBuffer response = new StringBuffer();
while ((output = in.readLine()) != null) {
response.append(output);
}
in.close();
System.out.println(response.toString());
}}
<a name="python"><br/></a>
### 3.2 MatrixGate HTTP API Python Exampleimport http.client
class MxgateExample(object): def init(self):
/*
* mxgate listens on port 8086 of localhost
*/
self.url = "localhost:8086"
self.postData = "public.testtable\n/" \
"1603777821|1|101|201|301\n/" \
"1603777822|2|102|202|302\n/" \
"1603777823|3|103|203|303"
self.headers = {"Content-Type": "text/plain"}
/*
* HTTP Post request
*/
def sending_post_request(self):
conn = http.client.HTTPConnection(self.url)
conn.request("POST", "/", self.postData, self.headers)
response = conn.getresponse()
response_code = response.getcode()
print(f"Sending 'POST' request to URL : {self.url}")
print(f"Post Data : {self.postData}")
print(f"Response Code : {response_code}")
output = response.read()
print(output)if name == 'main': gate_post = MxgateExample() gate_post.sending_post_request()
<a name="C#"><br/></a>
### 3.3 MatrixGate HTTP API C# Example
> It is recommended to use the C# Core development environment for developing codeusing System; using System.IO; using System.Net; using System.Text;
namespace HttpPostTest { class Program { static void Main(string[] args) { var url = "http://10.13.2.177:8086/"; var txt = "public.dest\n2021-01-01 00:00:00,1,a1\n2021-01-01 00:00:00,2,a2\n2021-01-01 00:00:00,3,a3";
HttpPost(url,txt);
}public static string HttpPost(string url, string content){ string result = ""; HttpWebRequest req = (HttpWebRequest)WebRequest.Create(url); req.Method = "POST"; req.ContentType = "text/plain";
/*
* region Add Post parameters
*/
byte[] data = Encoding.UTF8.GetBytes(content);
req.ContentLength = data.Length;
using (Stream reqStream = req.GetRequestStream()){
reqStream.Write(data, 0, data.Length);
reqStream.Close();
}
HttpWebResponse resp = (HttpWebResponse)req.GetResponse();
Stream stream = resp.GetResponseStream();
/*
* Get the response content
*/
using (StreamReader reader = new StreamReader(stream, Encoding.UTF8)){
result = reader.ReadToEnd();
}
return result;
}} }
> If you encounter an error when serving connection ***** body size exceeds the given limit, you need to increase the `max-body-bytes` under `mxgate.conf`.
### 3.4 MatrixGate HTTP API Golang Examplepackage main
import ( "bytes" "net/http" )
func PostDataToServer(URL string) error {
data := public.testtable 1603777821|1|101|201|301 1603777822|2|102|202|302 1603777823|3|103|203|303
resp, err := http.Post(URL, "application/text", bytes.NewBuffer([]byte(data)))
if err != nil {
return err
}
if resp.StatusCode != 200 {
/*
/*
* Deal with the response body.
*/
return nil}
func main() { err := PostDataToServer("http://127.0.0.1:8086") if err != nil{ panic(err) }
}
>***Note!***
For more MatrixGate features, see [Tool Guide - mxgate](/doc/5.1/tools/mxgate/feature).