Doris Writer
Doris Writer plugin writes data into Apache Doris through its native HTTP Stream Load interface (FE http_port, default 8030). Compared with insert into it is considerably faster and is the officially recommended way to load data in production.
Doris is MySQL protocol compatible, so it can be read with MySQL Reader.
Example
Assume the target table is created as follows:
CREATE DATABASE example_db;
CREATE TABLE example_db.table1
(
siteid INT DEFAULT '10',
citycode SMALLINT,
username VARCHAR(32) DEFAULT '',
pv BIGINT SUM DEFAULT '0'
) AGGREGATE KEY(siteid, citycode, username)
DISTRIBUTED BY HASH(siteid) BUCKETS 10
PROPERTIES("replication_num" = "1");The following configuration reads from memory and writes into the Doris table:
{
"job": {
"setting": {
"speed": {
"channel": 2
}
},
"content": {
"reader": {
"name": "streamreader",
"parameter": {
"column": [
{
"random": "1,500",
"type": "long"
},
{
"random": "1,127",
"type": "long"
},
{
"value": "this is a text",
"type": "string"
},
{
"random": "5,200",
"type": "long"
}
],
"sliceRecordCount": 100
}
},
"writer": {
"name": "doriswriter",
"parameter": {
"loadUrl": [
"127.0.0.1:8030"
],
"username": "test",
"password": "123456",
"batchSize": 1024,
"column": [
"siteid",
"citycode",
"username",
"pv"
],
"connection": {
"table": ["table1"],
"database": "example_db",
"jdbcUrl": "jdbc:mysql://localhost:9030/example_db"
},
"loadProps": {
"format": "json",
"strip_outer_array": true
}
}
}
}
}
}Save the configuration file as job/stream2doris.json and run it:
bin/addax.sh job/stream2doris.jsonThe output looks like:
Details
2021-02-23 15:22:57.851 [main] INFO VMInfo - VMInfo# operatingSystem class => sun.management.OperatingSystemImpl
2021-02-23 15:22:57.871 [main] INFO Engine -
{
"content":{
"reader":{
"parameter":{
"column":[
{
"random":"1,500",
"type":"long"
},
{
"random":"1,127",
"type":"long"
},
{
"type":"string",
"value":"username"
}
],
"sliceRecordCount":100
},
"name":"streamreader"
},
"writer":{
"parameter":{
"password":"*****",
"batchSize":1024,
"connection":[
{
"database":"example_db",
"endpoint":"http://127.0.0.1:8030/",
"table":"table1"
}
],
"username":"test"
},
"name":"doriswriter"
}
},
"setting":{
"speed":{
"channel":2
}
}
}
2021-02-23 15:22:57.886 [main] INFO PerfTrace - PerfTrace traceId=job_-1, isEnable=false, priority=0
2021-02-23 15:22:57.886 [main] INFO JobContainer - Addax jobContainer starts job.
2021-02-23 15:22:57.920 [job-0] INFO JobContainer - Scheduler starts [1] taskGroups.
2021-02-23 15:22:57.928 [taskGroup-0] INFO TaskGroupContainer - taskGroupId=[0] start [2] channels for [2] tasks.
2021-02-23 15:22:57.935 [taskGroup-0] INFO Channel - Channel set byte_speed_limit to -1, No bps activated.
2021-02-23 15:22:57.936 [taskGroup-0] INFO Channel - Channel set record_speed_limit to -1, No tps activated.
2021-02-23 15:22:57.970 [0-0-1-writer] INFO DorisWriterTask - connect DorisDB with http://127.0.0.1:8030//api/example_db/table1/_stream_load
2021-02-23 15:22:57.970 [0-0-0-writer] INFO DorisWriterTask - connect DorisDB with http://127.0.0.1:8030//api/example_db/table1/_stream_load
2021-02-23 15:23:00.941 [job-0] INFO JobContainer - PerfTrace not enable!
2021-02-23 15:23:00.946 [job-0] INFO JobContainer -
任务启动时刻 : 2021-02-23 15:22:57
任务结束时刻 : 2021-02-23 15:23:00
任务总计耗时 : 3s
任务平均流量 : 1.56KB/s
记录写入速度 : 66rec/s
读出记录总数 : 200
读写失败总数 : 0Parameters
| Item | Required | Type | Default | Description |
|---|---|---|---|---|
| loadUrl | yes | string | none | Stream Load connection target |
| username | yes | string | none | User name used to access the Doris database |
| password | no | string | none | Password used to access the Doris database |
| flushInterval | no | int | 3000 | How often buffered data is written to the target table, in milliseconds |
| flushQueueLength | no | int | 1 | Length of the queue holding the batches waiting to be uploaded |
| table | yes | List | none | The tables to be synchronized |
| column | yes | list | none | Columns to synchronize, see RBDMS Writer for details |
| batchSize | no | int | 2048 | Max rows of one batch; the batch is written as soon as this many rows are buffered |
| connectTimeout | no | int | 5000 | Connection timeout in milliseconds |
| socketTimeout | no | int | 600000 | Response timeout in milliseconds; must cover the time a full sized batch takes to load |
| connectionRequestTimeout | no | int | 5000 | Timeout for leasing a connection from the pool, in milliseconds |
| hostCooldownMs | no | int | 30000 | How long a loadUrl is skipped after a failed load, in milliseconds |
| loadProps | no | map | csv | Stream Load request parameters, see the StreamLoad page |
| preSql | no | list | SQL statements to execute before writing data into the target table | |
| postSql | no | list | SQL statements to execute after all data has been written |
loadUrl
The Stream Load connection target, in the form ip:port, where the IP is a FE node and the port is its http_port. Entries prefixed with http:// or https:// are accepted as well.
Multiple entries may be configured. The plugin then picks FE nodes in round-robin order; a node that failed a load is skipped for hostCooldownMs (30 seconds by default) so that later batches do not keep hitting a broken node. When every node is cooling down the plugin still issues the request against the round-robin candidate and lets the batch level retry move to another node, instead of failing right away.
column
Setting it to ["*"] makes the plugin read the column list from the Doris table and assemble the records accordingly.
loadProps
Stream Load request parameters, see Stream load - Apache Doris for the full list. This is also where the import format (format) is selected: csv is the default, json is supported as well.
Type Conversion
All values are converted to strings by default and joined with \t as the column separator and \n as the line delimiter, producing the CSV file uploaded by Stream Load.
To change the delimiters, configure loadProps:
{
"loadProps": {
"column_separator": "\\x01",
"line_delimiter": "\\x02"
}
}To import JSON instead, configure loadProps:
{
"loadProps": {
"format": "json",
"strip_outer_array": true
}
}Note: in CSV mode the plugin does not escape the field content, so a field containing the column separator (\t by default) or a newline breaks the row. Use format: json for such data, or replace the special characters upstream.
Performance tuning
Every Stream Load request creates one import transaction in Doris, so small batches produce a large number of small transactions, which adds version and compaction pressure. batchSize defaults to 2048 rows (the same default as the other writers in this project); raise it for jobs that write a lot of data.
Recommendations:
- Raising
batchSize(50000 to 200000 rows for example) cuts the number of import transactions substantially. Keep in mind that it counts rows: for wide tables with large rows, work out what that means in bytes so a single batch does not take up too much memory. flushIntervalis the longest time a record waits to be batched. Lower it when the write rate is low and latency matters; when the write rate is high the row limit triggers first, so raisebatchSizeinstead to get bigger batches.flushQueueLengthis the number of batches waiting to be uploaded and therefore also bounds the memory usage (roughly (queue length + 1) × bytes per batch). Keep the default of 1 when memory is tight.- Every task reuses one HTTP connection pool, connections are no longer re-established per batch. On a low latency network
connectTimeoutcan be lowered, whilesocketTimeouthas to cover the load time of the largest batch or a big batch will be reported as timed out. - While a job runs, the
rows[]andbytes[]values in the log show the actual batch sizes. - Note that measured end to end, the batch size only shifts the wall clock time as far as the Doris cluster and the network allow; on a slow load path the difference is small. The main benefit of a larger
batchSizeis fewer transactions and less compaction pressure.