Build a Real-Time Graph Analysis Pipeline from MySQL via DTS - PolarDB - Alibaba Cloud - PolarDB

Prerequisites

Version requirements

The following PolarDB for PostgreSQL versions are supported:

PostgreSQL 16 (revision version 2.0.16.8.3.0 or later)
PostgreSQL 15 (revision version 2.0.15.12.4.0 or later)
PostgreSQL 14 (revision version 2.0.14.12.24.0 or later)

Note

You can check the revision version in the console or by running the SHOW polardb_version; statement. If the revision version does not meet the requirements, update the revision version.

Data format

Both node and edge data require a column for a unique ID within their label type. The ID must be less than 2^48. In addition to a unique ID, edge data must also include columns for the start and end node IDs.
If a node or edge lacks a unique ID, or if the unique ID is not an integer, you can add a serial column to provide the unique ID. Without a unique ID, DTS cannot synchronize updates and deletions. You can use the serial feature to automatically generate the ID, and you can choose not to import this ID column into the graph.

How it works

Procedure

Prepare resources

A PolarDB for MySQL cluster as the data source. For more information, see Purchase a cluster.
A PolarDB for PostgreSQL cluster. For more information, see Create a database cluster.
The PolarDB for MySQL and PolarDB for PostgreSQL clusters must be in the same region, zone, and VPC.

Steps

Write new data to the PolarDB for MySQL cluster and use a DTS task to synchronize the data to the graph database in PolarDB for PostgreSQL.

Prepare the base data.

Assume the graph data to synchronize consists of three tables: two node tables, raw_a and raw_b, each with a unique ID, and an edge table, raw_c, which stores the start node ID from raw_a and the end node ID from raw_b. Then, create a graph in PolarDB for PostgreSQL that includes two node labels (label_a and label_b) and one edge label (edge_c).
- PolarDB for MySQL
  1. Create table definitions.
```
CREATE TABLE raw_a(id integer, name text, `desc` text, time_created timestamp);
CREATE TABLE raw_b(id integer, name text, `desc` text, `value` integer, time_created timestamp);
CREATE TABLE raw_c(id integer, id_a integer, id_b integer);
```
  2. Before you use DTS, you must enable the binary logging feature for PolarDB for MySQL. For more information, see Enable binary logging.
- PolarDB for PostgreSQL
  1. In the target database, use a privileged account to create and load the extension. To create a privileged account, see Create a database account.
    
    Note
    Compatibility issues may occur when you use Data Management (DMS) to configure the search_path. In such cases, you can use PolarDB-Tools to execute related statements.
```
CREATE EXTENSION age;
ALTER DATABASE <dbname> SET search_path = "$user", public, ag_catalog;
ALTER DATABASE <dbname> SET session_preload_libraries TO 'age';
```
  2. Create the graph, nodes, and edges.
```
SELECT create_graph('gra');
SELECT create_vlabel('gra', 'label_a');
SELECT create_vlabel('gra', 'label_b');
SELECT create_elabel('gra', 'edge_c');
```

Use DTS to synchronize data to PolarDB for PostgreSQL.

Note

Do not write data while creating the synchronization task. Otherwise, this data cannot be imported into the graph.
After the DTS task is created, do not modify the data structure of the synchronized tables, for example, by adding or deleting columns. Otherwise, subsequent synchronization may fail.

Go to the data synchronization task list page in the destination region. You can do this in one of two ways.
DTS console
1. Log on to the DTS console.
2. In the navigation pane on the left, click Data Synchronization.
3. In the upper-left corner of the page, select the region where the synchronization instance is located.
DMS console
Note
The actual steps may vary depending on the mode and layout of the DMS console. For more information, see Simple mode console and Customize DMS console layout and style.
1. Log on to the DMS console.
2. In the top menu bar, choose Data + AI > DTS (DTS) > Data Synchronization.
3. To the right of Data Synchronization Tasks, select the region of the synchronization instance.
Click Create Task to open the task configuration page.

On the task creation page, configure the following parameters.

Note

This topic describes only some of the configuration items. For more information about all parameters, see Configure a synchronization task.

Category	Setting	Description
Source Database	Database Type	Select MySQL.
Source Database	Connection Type	Select Express Connect, VPN Gateway, or Smart Access Gateway.
Destination Database	Database Type	Select PostgreSQL.
Destination Database	Connection Type	Select Express Connect, VPN Gateway, or Smart Access Gateway.

Enter information such as the instance region, VPC CIDR block, cluster endpoint, port, username, and password based on your cluster configuration.

In the Configure Task Objects step, in the Configure Objects section, select the three tables raw_a, raw_b, and raw_c from the TABLE column for the database. Use the default configurations for the subsequent steps and click Next.

Note
For more information about how to configure task objects, see Configure task objects and Advanced settings.
Save the task and run a precheck. When the Success Rate is 100%, click Next: Purchase Instance.
On the Purchase page, select an instance class and billing method for the data synchronization instance. Then, read and accept the Data Transmission Service (Pay-as-you-go) Service Terms and click Buy and Start. In the OK dialog box, click OK.
Note
- For more information about how to configure the parameters on the Purchase page, see Purchase a data synchronization instance.
- You can view the task progress on the DTS data synchronization page.

Synchronize data to the graph using triggers.

In the PolarDB for PostgreSQL cluster, create the following auxiliary functions:

--- Function 1
CREATE OR REPLACE FUNCTION age_name_to_idx_start(graph_name text, kind_name text, label_name text)
RETURNS bigint
AS 'SELECT id::bigint<<48 FROM ag_catalog.ag_label WHERE kind = kind_name and name = label_name and graph = (SELECT graphid FROM ag_catalog.ag_graph WHERE name = graph_name)'
language SQL IMMUTABLE STRICT PARALLEL SAFE;



--- Function 2
CREATE OR REPLACE FUNCTION build_age_triggers_for_vertex(schema_name text, table_name text, table_id_col text, graph_name text, graph_label text)
RETURNS BOOL
AS
$outer$
DECLARE
  column_names TEXT;
  sql TEXT;
BEGIN
  SELECT string_agg(format('val.%I', column_name), ', ')
    INTO column_names
    FROM information_schema.columns
    WHERE columns.table_schema = build_age_triggers_for_vertex.schema_name AND columns.table_name = build_age_triggers_for_vertex.table_name;
  sql := $$

CREATE OR REPLACE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_id(id bigint)
RETURNS graphid
AS 'SELECT (age_name_to_idx_start(''$$ || graph_name || $$'', ''v'', ''$$ || graph_label|| $$'') + id)::text::ag_catalog.graphid'
LANGUAGE SQL;

CREATE OR REPLACE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_properties(val $$ || schema_name || $$.$$ || table_name || $$)
RETURNS agtype
AS 'SELECT row_to_json((select x FROM (select $$|| column_names || $$) x))::text::agtype'
LANGUAGE SQL;

CREATE OR REPLACE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$() RETURNS TRIGGER AS
$inner$
BEGIN
  IF TG_OP = 'INSERT' THEN
    INSERT INTO "$$ || graph_name || $$"."$$ || graph_label || $$" (id, properties) VALUES (_sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_id(NEW."$$ || table_id_col || $$"), _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_properties(NEW));
    RETURN NEW;
  ELSIF TG_OP = 'UPDATE' THEN
    UPDATE "$$ || graph_name || $$"."$$ || graph_label || $$" SET properties = _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_properties(NEW) WHERE id = _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_id(OLD."$$ || table_id_col || $$");
    RETURN NEW;
  ELSIF TG_OP = 'DELETE' THEN
    DELETE FROM "$$ || graph_name || $$"."$$ || graph_label || $$" WHERE id = _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_id(OLD."$$ || table_id_col || $$");
    RETURN OLD;
  END IF;
  RETURN NULL;
END;
$inner$ LANGUAGE plpgsql;

CREATE OR REPLACE TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_insert
AFTER INSERT ON $$ || schema_name || $$.$$ || table_name || $$
FOR EACH ROW EXECUTE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$();

CREATE OR REPLACE TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_update
AFTER UPDATE ON $$ || schema_name || $$.$$ || table_name || $$
FOR EACH ROW EXECUTE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$();

CREATE OR REPLACE TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_delete
AFTER DELETE ON $$ || schema_name || $$.$$ || table_name || $$
FOR EACH ROW EXECUTE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$();

ALTER TABLE $$ || schema_name || $$.$$ || table_name || $$ ENABLE ALWAYS TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_insert;
ALTER TABLE $$ || schema_name || $$.$$ || table_name || $$ ENABLE ALWAYS TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_update;
ALTER TABLE $$ || schema_name || $$.$$ || table_name || $$ ENABLE ALWAYS TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_delete;
  $$;
  EXECUTE sql;
  RETURN true;
END;
$outer$
LANGUAGE plpgsql;



--- Function 3
CREATE OR REPLACE FUNCTION build_age_triggers_for_edge(schema_name text, table_name text, table_id_col text, start_table_name text, start_id_col text, end_table_name text, end_id_col text, graph_name text, graph_label text)
RETURNS BOOL
AS
$outer$
DECLARE
  column_names TEXT;
  sql TEXT;
BEGIN
  SELECT string_agg(format('val.%I', column_name), ', ')
    INTO column_names
    FROM information_schema.columns
    WHERE columns.table_schema = build_age_triggers_for_edge.schema_name AND columns.table_name = build_age_triggers_for_edge.table_name;
  sql := $$

CREATE OR REPLACE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_id(id bigint)
RETURNS graphid
AS 'SELECT (age_name_to_idx_start(''$$ || graph_name || $$'', ''e'', ''$$ || graph_label|| $$'') + id)::text::ag_catalog.graphid'
LANGUAGE SQL;

CREATE OR REPLACE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_properties(val $$ || schema_name || $$.$$ || table_name || $$)
RETURNS agtype
AS 'SELECT row_to_json((select x FROM (select $$|| column_names || $$) x))::text::agtype'
LANGUAGE SQL;

CREATE OR REPLACE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$() RETURNS TRIGGER AS
$inner$
BEGIN
  IF TG_OP = 'INSERT' THEN
    INSERT INTO "$$ || graph_name || $$"."$$ || graph_label || $$" (id, start_id, end_id, properties) VALUES (_sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_id(NEW."$$ || table_id_col || $$"), _sync_$$ || schema_name || $$_$$ || start_table_name || $$_row_to_id(NEW."$$ || start_id_col || $$"), _sync_$$ || schema_name || $$_$$ || end_table_name || $$_row_to_id(NEW."$$ || end_id_col || $$"), _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_properties(NEW));
    RETURN NEW;
  ELSIF TG_OP = 'UPDATE' THEN
    UPDATE "$$ || graph_name || $$"."$$ || graph_label || $$" SET start_id = _sync_$$ || schema_name || $$_$$ || start_table_name || $$_row_to_id(NEW."$$ || start_id_col || $$"), end_id = _sync_$$ || schema_name || $$_$$ || end_table_name || $$_row_to_id(NEW."$$ || end_id_col || $$"), properties = _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_properties(NEW) WHERE id = _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_id(OLD."$$ || table_id_col || $$");
    RETURN NEW;
  ELSIF TG_OP = 'DELETE' THEN
    DELETE FROM "$$ || graph_name || $$"."$$ || graph_label || $$" WHERE id = _sync_$$ || schema_name || $$_$$ || table_name || $$_row_to_id(OLD."$$ || table_id_col || $$");
    RETURN OLD;
  END IF;
  RETURN NULL;
END;
$inner$ LANGUAGE plpgsql;

CREATE OR REPLACE TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_insert
AFTER INSERT ON $$ || schema_name || $$.$$ || table_name || $$
FOR EACH ROW EXECUTE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$();

CREATE OR REPLACE TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_update
AFTER UPDATE ON $$ || schema_name || $$.$$ || table_name || $$
FOR EACH ROW EXECUTE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$();

CREATE OR REPLACE TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_delete
AFTER DELETE ON $$ || schema_name || $$.$$ || table_name || $$
FOR EACH ROW EXECUTE FUNCTION _sync_$$ || schema_name || $$_$$ || table_name || $$();

ALTER TABLE $$ || schema_name || $$.$$ || table_name || $$ ENABLE ALWAYS TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_insert;
ALTER TABLE $$ || schema_name || $$.$$ || table_name || $$ ENABLE ALWAYS TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_update;
ALTER TABLE $$ || schema_name || $$.$$ || table_name || $$ ENABLE ALWAYS TRIGGER _sync_$$ || schema_name || $$_$$ || table_name || $$_delete;
  $$;

  EXECUTE sql;
  RETURN true;
END;
$outer$
LANGUAGE plpgsql;

Execute the auxiliary functions to create triggers that synchronize data from the tables to the graph.
Note
- Use only lowercase letters. Names are case-sensitive.
- Replace your_schema_name with the name of the schema that contains tables such as raw_a. You can run the \d+ <table_name> command in the psql client to view the schema name. It is usually the same as the schema of the original table.
```
select build_age_triggers_for_vertex('your_schema_name','raw_a', 'id', 'gra', 'label_a');
select build_age_triggers_for_vertex('your_schema_name','raw_b', 'id', 'gra', 'label_b');
select build_age_triggers_for_edge('your_schema_name','raw_c', 'id', 'raw_a', 'id_a', 'raw_b', 'id_b', 'gra', 'edge_c');
```

Note

These triggers synchronize only incremental data. To import existing data, run the following statements after creating the triggers:

INSERT INTO "gra"."label_a" (id, properties) SELECT _sync_your_schema_name_raw_a_row_to_id(t.id), _sync_your_schema_name_raw_a_row_to_properties(t) FROM your_schema_name.raw_a t;
INSERT INTO "gra"."label_b" (id, properties) SELECT _sync_your_schema_name_raw_b_row_to_id(t.id), _sync_your_schema_name_raw_b_row_to_properties(t) FROM your_schema_name.raw_b t;
INSERT INTO "gra"."edge_c" (id, start_id, end_id, properties) SELECT _sync_your_schema_name_raw_c_row_to_id(t.id), _sync_your_schema_name_raw_a_row_to_id(t.id_a), _sync_your_schema_name_raw_b_row_to_id(t.id_b), _sync_your_schema_name_raw_c_row_to_properties(t) FROM your_schema_name.raw_c t;

Test and verify.

Data insertion

In PolarDB for MySQL, insert test data into the tables to be synchronized.

INSERT INTO raw_a values(1,1,1,'2000-01-01');
INSERT INTO raw_b values(1,1,1,1,'2000-01-01');
INSERT INTO raw_c values(1,1,1);

In the PolarDB for PostgreSQL cluster, use Cypher to run a graph query and verify that the data was successfully inserted:

SELECT * FROM cypher('gra', $$
MATCH (v)
RETURN v
$$) as (v agtype);

The following result is returned:

------
 {"id": 844424930131969, "label": "label_a", "properties": {"id": 1, "desc": "1", "name": "1", "time_created": "2000-01-01T00:00:00"}}::vertex
 {"id": 1125899906842625, "label": "label_b", "properties": {"id": 1, "desc": "1", "name": "1", "value": 1, "time_created": "2000-01-01T00:00:00"}}::vertex

SELECT * FROM cypher('gra', $$
MATCH (v)-[e]->(v2)
RETURN e
$$) as (e agtype);

The following result is returned:

------
 {"id": 1407374883553281, "label": "edge_c", "end_id": 1125899906842625, "start_id": 844424930131969, "properties": {"id": "11"}}::edge

Property modification

In PolarDB for MySQL, update a property in the tables to be synchronized.
```
UPDATE raw_a SET name = '2' WHERE id = 1;
```

In the PolarDB for PostgreSQL cluster, run a graph query in Cypher to confirm that the update succeeded:

SELECT * FROM cypher('gra', $$
MATCH (v:label_a {id:1})
RETURN v
$$) as (v agtype);

The following result is returned:

-----
 {"id": 844424930131969, "label": "label_a", "properties": {"id": 1, "desc": "1", "name": "2", "time_created": "2000-01-01T00:00:00"}}::vertex

Data deletion

In PolarDB for MySQL, delete the test data from the tables to be synchronized.
```
DELETE FROM raw_c WHERE id = 1;
```
In the PolarDB for PostgreSQL cluster, run a graph query in Cypher to confirm that the deletion succeeded:
```
SELECT * FROM cypher('gra', $$
MATCH (v)-[e]->(v2)
RETURN e
$$) as (e agtype);
```
An empty result set indicates that the deletion has been synchronized.

Usage notes

As a reminder, do not write data during the initial task creation. This data will not be synchronized to the graph.
Avoid altering the table structure (for example, by adding or deleting columns) after the DTS task is running, as this can cause synchronization to fail.
By default, the provided auxiliary functions add all table columns as graph properties. To customize this behavior, modify the function definition named _sync_<table_name>_row_to_properties. In this function, you can change which columns are included or how their values are transformed by updating the SELECT clause. For example, to concatenate the id_a and id_b columns, change the clause to select val.id_a::text || val.id_b::text AS id.
```
CREATE OR REPLACE FUNCTION _sync_raw_C_row_to_properties(val raw_C)
RETURNS agtype
AS 'SELECT row_to_json((select x FROM (select val.id, val.id_a, val.id_b) x))::text::agtype'
LANGUAGE SQL;
```
When inserting an edge, ensure both its start and end nodes already exist in the graph. Otherwise, queries involving that edge may fail.

Prerequisites

Version requirements

Data format

How it works

Procedure

Prepare resources

Steps

DTS console

DMS console

Data insertion

Property modification

Data deletion

Usage notes