数据加载服务器MatrixGate

MatrixGate简称mxgate，是高性能流式数据加载服务器，位于MatrixDB安装目录下的bin/mxgate。MatrixGate当前提供HTTP和STDIN接口加载数据，数据格式支持TEXT和CSV。

1 MatrixGate工作原理

MatrixGate加载数据的逻辑如下图所示，1）数据采集系统采集设备数据或者接收由设备发送来的数据 2) 采集系统以并发微批的模式向MatrixGate的服务进程mxgate持续发送数据 3）mxgate进程和MatrixDB的master进程高效通信，沟通事务和控制信息 4）数据直接发送到segment节点，并行高速写入。

2 MatrixGate用法

• 指定目标数据库和目标表，生成mxgate配置文件

  mxgate config --db-database demo --target public.testtable --target public.testtable2 --allow-dynamic > mxgate.conf

  上述参数将生成一个配置文件mxgate.conf，允许用户对testtable和testtable2的加载做些个性化定制，同时也可以使用全局缺省设置往其他表中加载数据。

• 根据需要修改mxgate配置文件，如配置数据分隔符等，选择默认配置可忽略此步骤。可以在该配置文件中看到与testtable和testtable2对应的设置如下：

    [[job.target]]
      # delimiter = "|"
      # exclude-columns = []
      # format = "text"
      name = "job_text_to_public.testtable"
      # null-as = ""
      table = "public.testtable"
      # time-format = "unix-second"
      # use-auto-increment = true
  
    [[job.target]]
      # delimiter = "|"
      # exclude-columns = []
      # format = "text"
      name = "job_text_to_public.testtable2"
      # null-as = ""
      table = "public.testtable2"
      # time-format = "unix-second"
      # use-auto-increment = true

  如果testtable的分隔符是@，而testtable2的分隔符是%，可以把上述配置修改为：

    [[job.target]]
      delimiter = "@"
      # exclude-columns = []
      # format = "text"
      name = "job_text_to_public.testtable"
      # null-as = ""
      table = "public.testtable"
      # time-format = "unix-second"
      # use-auto-increment = true
  
    [[job.target]]
      delimiter = "%"
      # exclude-columns = []
      # format = "text"
      name = "job_text_to_public.testtable2"
      # null-as = ""
      table = "public.testtable2"
      # time-format = "unix-second"
      # use-auto-increment = true

  mxgate默认监听8086端口接收数据，可以在mxgate.conf中看到source.http下的http-port子项设置为8086，如果需要，可以改为其他端口：

  [source]
  
  ## Source plugin is the data entrance to MatrixGate
  ## Types restricted to: http
  source = "http"
  
  [source.http]
  
    ## Port of http push
    # http-port = 8086
  
    ## Maximum request body size (after gzip)
    ## The server rejects requests with bodies exceeding this limit.
    # max-body-bytes = 4194304
  
    ## The maximum number of concurrent HTTP connections to the server
    ## The server response with 503 after exceed this limit.
    # max-concurrency = 40000

• 启动mxgate，加载配置文件，连接demo数据库，准备接收数据加载请求

  mxgate start --config mxgate.conf

• 查看后台服务状态

  mxgate status

• 终止后台服务

  mxgate stop

  当遇到超时或者其他问题需要强制停止时，可以这样执行：

  mxgate stop --force

3 MatrixGate命令行参数详解

4 MatrixGate API

MatrixGate对外提供HTTP API，支持各种编程语言通过HTTP接口将数据导入到MatrixDB数据库中。

MatrixGate HTTP协议格式

MatrixGate HTTP响应码

5 MatrixGate HTTP API 命令行示例

• 在demo数据库中创建表testtable

  CREATE TABLE testtable (time TIMESTAMP WITH TIME ZONE, tagid INT, c1 INT, c2 INT, c3 INT)
  DISTRIBUTED BY (tagid);

• 编辑数据加载文件data.txt，内容如下

  public.testtable
  1603777821|1|101|201|301
  1603777822|2|102|202|302
  1603777823|3|103|203|303

• 启动mxgate，指定生成好的配置文件mxgate.conf

  mxgate --config mxgate.conf

• 发送HTTP请求加载数据

  curl http://localhost:8086/ -X POST -H 'Content-Type: text/plain' --data-binary "@data.txt"

• 连接数据库查询数据是否加载成功

demo=# SELECT * FROM testtable ;
          time          | tagid | c1  | c2  | c3
------------------------+-------+-----+-----+-----
 2020-10-27 05:50:23+08 |     3 | 103 | 203 | 303
 2020-10-27 05:50:22+08 |     2 | 102 | 202 | 302
 2020-10-27 05:50:21+08 |     1 | 101 | 201 | 301

 (3 rows)

6 编程语言连接MatrixGate

6.1 MatrixGate HTTP API Java示例

import 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监听在localhost的8086端口
        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());
        // 数据有中文时，可以通过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());
    }
}

6.2 MatrixGate HTTP API Python示例

import http.client

class MxgateExample(object):
    def __init__(self):
        # mxgate监听在localhost的8086端口
        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()

6.3 MatrixGate HTTP API C#示例

建议开发代码使用C# Core 开发环境

using 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 添加Post 参数
    byte[] data = Encoding.UTF8.GetBytes(content);
    req.ContentLength = data.Length;
    using (Stream reqStream = req.GetRequestStream()){
        reqStream.Write(data, 0, data.Length);
        reqStream.Close();
    }
    #endregion

    HttpWebResponse resp = (HttpWebResponse)req.GetResponse();
    Stream stream = resp.GetResponseStream();
    //获取响应内容
    using (StreamReader reader = new StreamReader(stream, Encoding.UTF8)){
        result = reader.ReadToEnd();
        }
        return result;
    }

  }
}

如果遇到error when serving connection ***** body size exceeds the given limit 这个问题，调大mxgate.conf 下的max-body-bytes

6.4 MatrixGate HTTP API Golang示例

package 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 }

// Deal with the response body.
return nil

}

func main() { err := PostDataToServer("http://127.0.0.1:8086") if err != nil{ panic(err) }

}

## 7 MatrixGate加载特殊类型
### 7.1 MatrixGate 加载CSV文件示例
- 在demo数据库中创建表csvtable

CREATE TABLE csvtable (time TIMESTAMP WITH TIME ZONE, tagid INT, c1 INT, c2 INT, c3 INT) DISTRIBUTED BY (tagid);

- 编辑数据加载文件data.csv，内容如下

1603777821|1|101|201|301 1603777822|2|102|202|302 1603777823|3|103|203|303

- 启动mxgate，指定source参数为stdin，目标表为已经存在的csvtable，加载并行度为2

mxgate \ --source stdin \ --db-database demo \ --db-master-host 127.0.0.1 \ --db-master-port 5432 \ --db-user mxadmin \ --time-format unix-second \ --delimiter "|" \ --target csvtable \ --parallel 2 < data.csv

- 连接数据库查询数据是否加载成功

demo=# SELECT * FROM csvtable ; time | tagid | c1 | c2 | c3 ------------------------+-------+-----+-----+----- 2020-10-27 05:50:23+08 | 3 | 103 | 203 | 303 2020-10-27 05:50:22+08 | 2 | 102 | 202 | 302 2020-10-27 05:50:21+08 | 1 | 101 | 201 | 301

(3 rows)

### 7.2 MatrixGate 加载json字段示例

### 7.2.1 json
- 创建表

create table json_test(id int, j json);

- 创建数据文件
`~/json.csv`

1|"{""a"":10, ""b"":""xyz""}"

- 加载
这里使用stdin模式为例，其他模式都一样。
关键在 `--format csv`

mxgated \ --source stdin \ --db-database postgres \ --db-master-host 127.0.0.1 \ --db-master-port 7000 \ --db-user mxadmin \ --time-format raw \ --format csv \ --delimiter "|" \ --target json_test < ~/json.csv

- 查看加载数据

postgres=# select * from json_test; id | j ----+----------------------- 1 | {"a":10, "b":"xyz"} (1 row)

### 7.2.2 json数组

- 创建表

create table json_array_test(id int, j _json);

- 创建数据文件
`~/json_array.csv`

1|"{""{\""a\"":10, \""b\"":\""xyz\""}"",""{\""c\"": 10}""}"

- 加载

mxgate \ --source stdin \ --db-database postgres \ --db-master-host 127.0.0.1 \ --db-master-port 7000 \ --db-user mxadmin \ --time-format raw \ --format csv \ --delimiter "|" \ --target json_array_test < ~/json_array.csv

- 验证

postgres=# select * from json_array_test ; id | j ----+--------------------------------------------- 1 | {"{\"a\":10, \"b\":\"xyz\"}","{\"c\": 10}"} (1 row)

> 注意：因为json列包含引号等特殊字符，所以mxgate的--format参数必须为csv