Migrate data from Doris to AnalyticDB for PostgreSQL

Preparations

• You have activated an Express Connect circuit for your on-premises data center. For more information, see Connect an on-premises data center to a VPC over an Express Connect circuit.

• You have activated Alibaba Cloud Object Storage Service (OSS). For more information, see What is OSS?.

• You have created an OSS bucket. For more information, see Create a bucket.

• You have created an AnalyticDB for PostgreSQL instance. For more information, see Create an instance.

Procedure

Step 1: Create a destination table to load data

In your AnalyticDB for PostgreSQL instance, create a destination table to load data from Doris. The schema of the destination table must match the schema of the source table. For more information about the table creation syntax, see CREATE TABLE statements.

Step 2: Import data from Doris to OSS

Doris can export data using the S3 protocol. Because Object Storage Service (OSS) is compatible with the S3 protocol, you can directly export data from Doris to OSS. However, the export capabilities vary based on the Doris version:

• Doris 2.0 and later support exporting data using the EXPORT TABLE statement. The supported formats are CSV, text, ORC, and Parquet. Data exported in the Parquet format is not compatible with the mainstream Parquet format. The CSV and text formats do not support exporting data that contains special characters, such as line feeds. Therefore, the ORC format is recommended because it provides faster export speeds.

• For Doris 1.2, use the SELECT INTO OUTFILE statement. In this version, the EXPORT TABLE statement supports only the CSV format, which has limitations. The SELECT INTO OUTFILE statement exports data in the ORC format. Although it is slightly slower than the EXPORT TABLE statement, it supports filtering with a WHERE clause.

Examples

The following code shows the export statements for Doris 2.0 and Doris 1.2:

------ Doris 2.0
EXPORT TABLE s3_test TO "s3://bucket/dir/" 
PROPERTIES (
  "format"="orc"
)
WITH s3 (
    "AWS_ENDPOINT" = "oss-cn-shanghai-internal.aliyuncs.com",
    "AWS_ACCESS_KEY" = "************************",
    "AWS_SECRET_KEY" = "************************",
    "AWS_REGION" = "shanghai"
)
---- Doris 1.2 The c column is in datetime format
SELECT a, b, CAST(c AS string) AS c FROM s3_test INTO OUTFILE "s3://bucket/dir/"
FORMAT AS orc
PROPERTIES
(
    "AWS_ENDPOINT" = "oss-cn-shanghai-internal.aliyuncs.com",
    "AWS_ACCESS_KEY" = "************************",
    "AWS_SECRET_KEY" = "************************",
    "AWS_REGION" = "shanghai"
);

Step 3: Import data from OSS to AnalyticDB for PostgreSQL

You can import data into AnalyticDB for PostgreSQL using the COPY command or an OSS foreign table:

• To import data from OSS using the COPY command, see Use the COPY or UNLOAD command to import data from or export data to OSS.

• To import data from OSS using an OSS foreign table, see Use an OSS foreign table for data lake analytics.

Example

The following code shows the statement for importing data from OSS using the COPY command:

COPY test1 FROM 'oss://bucket/dir/' ACCESS_KEY_ID '************************' SECRET_ACCESS_KEY '************************' 
FORMAT AS orc ENDPOINT 'oss-*****-
internal.aliyuncs.com' FDW 'oss_fdw' ;

Syntax transform

Data types

CREATE TABLE statements

The following sections describe several common models for CREATE TABLE statements.

Model 1: Detail model

The detail model has no restrictions on primary keys or aggregated columns. The DUPLICATE KEY in the CREATE TABLE statement specifies the columns by which the underlying data is sorted. In AnalyticDB for PostgreSQL, this corresponds to an AOCS or BEAM table. You can use the ORDER BY clause to specify the sort keys. You can enable AUTOMERGE to automatically sort data at regular intervals.

Example

CREATE TABLE IF NOT EXISTS example_tbl_by_default
(
    `timestamp` DATETIME NOT NULL COMMENT "Log time",
    `type` INT NOT NULL COMMENT "Log type",
    `error_code` INT COMMENT "Error code",
    `error_msg` VARCHAR(1024) COMMENT "Detailed error message",
    `op_id` BIGINT COMMENT "Owner ID",
    `op_time` DATETIME COMMENT "Processing time"
)
DUPLICATE KEY (`timestamp`,`type`,`error_code`)
DISTRIBUTED BY HASH(`type`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);

----AnalyticDB for PostgreSQL
CREATE TABLE IF NOT EXISTS example_tbl_by_default
(
    "timestamp" TIMESTAMP NOT NULL ,
    "type" INT NOT NULL ,
    error_code INT ,
    error_msg VARCHAR(1024),
    op_id BIGINT,
    op_time TIMESTAMP
)
WITH(appendonly = true, orientation = column)
DISTRIBUTED BY("type")
ORDER BY("timestamp","type",error_code);

COMMENT ON COLUMN example_tbl_by_default.timestamp IS 'Log time';

Model 2: Primary key model

The primary key model ensures the uniqueness of the primary key. You can use UNIQUE KEY to specify the uniqueness constraint. In AnalyticDB for PostgreSQL, this corresponds to a heap table. You can use PRIMARY KEY to specify the unique key.

Example

CREATE TABLE IF NOT EXISTS example_tbl_unique
(
    `user_id` LARGEINT NOT NULL COMMENT "User ID",
    `username` VARCHAR(50) NOT NULL COMMENT "User nickname",
    `city` VARCHAR(20) COMMENT "City where the user is located",
    `age` SMALLINT COMMENT "User age",
    `sex` TINYINT COMMENT "User gender",
    `phone` LARGEINT COMMENT "User phone number",
    `address` VARCHAR(500) COMMENT "User address",
    `register_time` DATETIME COMMENT "User registration time"
)
UNIQUE KEY(`user_id`, `username`)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);

----AnalyticDB for PostgreSQL
CREATE TABLE IF NOT EXISTS example_tbl_unique
(
    user_id BIGINT NOT NULL,
    username VARCHAR(50) NOT NULL,
    city VARCHAR(20),
    age SMALLINT,
    sex SMALLINT,
    phone BIGINT,
    address VARCHAR(500),
    register_time TIMESTAMP,
    PRIMARY KEY (user_id, username)
)
DISTRIBUTED BY (user_id);

COMMENT ON COLUMN example_tbl_unique.user_id IS 'User ID';

Model 3: Aggregation model

When you import data using the aggregation model, rows that have the same Aggregate Key columns are aggregated into a single row. The Value columns are aggregated based on the specified AggregationType. In AnalyticDB for PostgreSQL, this corresponds to a heap table. You can create a unique index on the Aggregate Key and use the UPSERT method to insert data. For more information, see Use INSERT ON CONFLICT to overwrite data.

Example

CREATE TABLE IF NOT EXISTS example_tbl_agg1
(
    `user_id` LARGEINT NOT NULL COMMENT "User ID",
    `date` DATE NOT NULL COMMENT "Date and time of data import",
    `city` VARCHAR(20) COMMENT "City where the user is located",
    `age` SMALLINT COMMENT "User age",
    `sex` TINYINT COMMENT "User gender",
    `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "Time of the user's last visit",
    `cost` BIGINT SUM DEFAULT "0" COMMENT "Total user spending",
    `max_dwell_time` INT MAX DEFAULT "0" COMMENT "Maximum user dwell time",
    `min_dwell_time` INT MIN DEFAULT "99999" COMMENT "Minimum user dwell time"
)
AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);

-----AnalyticDB for PostgreSQL does not support automatic pre-aggregation
CREATE TABLE IF NOT EXISTS example_tbl_agg1
(
    user_id BIGINT NOT NULL,
    "date" DATE NOT NULL,
    city VARCHAR(20),
    age SMALLINT,
    sex SMALLINT,
    last_visit_date TIMESTAMP DEFAULT '1970-01-01 00:00:00',
    cost BIGINT DEFAULT 0,
    max_dwell_time INT DEFAULT 0,
    min_dwell_time INT DEFAULT 99999,
    UNIQUE (user_id, "date", city, age, sex)
)
DISTRIBUTED BY(user_id);

INSERT INTO example_tbl_agg1 VALUES (10000,'2024-08-22','beijing', 18, 0, '2024-08-22 12:00:00', 20, 1000, 1000) ON CONFLICT (user_id, "date", city, age, sex) DO UPDATE SET last_visit_date = excluded.last_visit_date, cost = example_tbl_agg1.cost + excluded.cost, max_dwell_time = GREATEST(example_tbl_agg1.max_dwell_time, excluded.max_dwell_time), min_dwell_time = LEAST(example_tbl_agg1.min_dwell_time, excluded.min_dwell_time);

Partitions and bucketing

Doris uses PARTITION BY for partitioning and DISTRIBUTED BY for bucketing. The BUCKETS clause specifies the number of buckets. In AnalyticDB for PostgreSQL, PARTITION BY corresponds to the partition key and DISTRIBUTED BY corresponds to the distribution key.

Example

CREATE TABLE IF NOT EXISTS example_range_tbl
(
    `user_id` LARGEINT NOT NULL COMMENT "User ID",
    `date` DATE NOT NULL COMMENT "Date and time of data import",
    `timestamp` DATETIME NOT NULL COMMENT "Timestamp of data import",
    `city` VARCHAR(20) COMMENT "City where the user is located",
    `age` SMALLINT COMMENT "User age",
    `sex` TINYINT COMMENT "User gender",
    `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "Time of the user's last visit",
    `cost` BIGINT SUM DEFAULT "0" COMMENT "Total user spending",
    `max_dwell_time` INT MAX DEFAULT "0" COMMENT "Maximum user dwell time",
    `min_dwell_time` INT MIN DEFAULT "99999" COMMENT "Minimum user dwell time"
)
ENGINE=OLAP
AGGREGATE KEY(`user_id`, `date`, `timestamp`, `city`, `age`, `sex`)
PARTITION BY RANGE(`date`)
(
    PARTITION `p201701` VALUES LESS THAN ("2017-02-01"),
    PARTITION `p201702` VALUES LESS THAN ("2017-03-01"),
    PARTITION `p201703` VALUES LESS THAN ("2017-04-01"),
    PARTITION `p2018` VALUES [("2018-01-01"), ("2019-01-01"))
)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 16
PROPERTIES
(
    "replication_num" = "1"
);

----AnalyticDB for PostgreSQL
CREATE TABLE IF NOT EXISTS example_range_tbl
(
    user_id BIGINT NOT NULL,
    "date" DATE NOT NULL,
    city VARCHAR(20),
    age SMALLINT,
    sex SMALLINT,
    visit_date TIMESTAMP DEFAULT '1970-01-01 00:00:00',
    a_cost BIGINT DEFAULT 0,
    dwell_time INT DEFAULT 0
)
PARTITION BY RANGE("date")
(
    PARTITION p201701 VALUES START ("2017-02-01") INCLUSIVE,
    PARTITION p201702 VALUES START ("2017-03-01") INCLUSIVE,
    PARTITION p201703 VALUES START ("2017-04-01") INCLUSIVE,
    PARTITION p2018 VALUES START ("2018-01-01") INCLUSIVE END ("2019-01-01") EXCLUSIVE
)
DISTRIBUTED BY (user_id);