Synchronize data to OSS using cross-database Spark SQL

Procedure

1. Prepare the data and environment.

2. Create a cross-database Spark SQL node.

3. Configure the cross-database Spark SQL node.

4. Run and publish the cross-database Spark SQL node.

Preparations

Create an order table and a product table in your online MySQL database.

1. Log in to DMS 5.0.

2. Move the pointer over the icon in the upper-left corner of the DMS console and choose All Features > SQL Console > SQL Console.

   Note

   If you use the DMS console in normal mode, choose SQL Console > SQL Console in the top navigation bar.

3. In the Please select the database first dialog box, search for and select your MySQL database, and then click Confirm.

4. Create the t_order and t_product tables in your MySQL database.

   1. Create the t_order table. Paste the following SQL statement into the SQL editor and click Execute.

      SQL statement for creating the t_order table:

      CREATE TABLE `t_order` (
        `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'The primary key.',
        `product_id` bigint(20) NOT NULL COMMENT 'The product ID.',
        `gmt_create` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'The creation time.',
        `gmt_modified` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'The modification time.',
        `customer_id` bigint(20) NOT NULL COMMENT 'The customer ID.',
        `price` decimal(14,2) NOT NULL COMMENT 'The price.',
        `status` varchar(64) NOT NULL COMMENT 'The order status.',
        `province` varchar(256) DEFAULT NULL COMMENT 'The transaction province.',
        PRIMARY KEY (`id`),
        KEY `idx_product_id` (`product_id`),
        KEY `idx_customer_id` (`customer_id`),
        KEY `idx_status` (`status`)
      ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 ROW_FORMAT=COMPACT COMMENT='The order table.'
      ;

   2. Create the t_product table. Paste the following SQL statement into the SQL editor and click Execute.

      SQL statement for creating the t_product table:

      CREATE TABLE `t_product` (
        `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'The primary key.',
        `name` varchar(128) NOT NULL COMMENT 'The product name.',
        `type` varchar(64) NOT NULL COMMENT 'The product category.',
        `gmt_create` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'The creation time.',
        `gmt_modified` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'The modification time.',
        PRIMARY KEY (`id`)
      ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='The product table.'
      ;

5. Insert test data into the t_order and t_product tables. Use the test data generation feature. For more information, see Generate test data.

   • Generate 20 million rows of data for the t_order table.

     Note

     For instances in flexible management mode, each ticket can generate up to 1 million rows of data.

   • Generate 10,000 rows of data for the t_product table.

Create a cross-database Spark SQL node

1. Log in to DMS 5.0.

2. Move the pointer over the icon in the upper-left corner and choose All Features > Data+AI > Data Development > Task Orchestration.

   Note

   If you use the DMS console in normal mode, choose Data+AI > Data Development > Task Orchestration in the top navigation bar.

3. Create a task flow.

   1. Click Create task flow.

   2. Select a Business scenario for the task flow, enter a Task flow name and Description, and then click Confirm.

4. From the Task type list on the left side of the canvas, drag the Cross-database Spark SQL node to a blank area on the canvas.

Configure the Spark SQL node

1. On the task flow details page, double-click the Cross-database Spark SQL node.

2. On the configuration page, configure the ${today} variable to represent the current date. For more information about variables, see Variables.

   1. In the right-side pane, click Variable setting.

   2. On the Node variable tab, enter the variable details.

      A node variable is a time-based variable usable only within the current node, using the format ${name}. For example, set the variable name to today, the time format to yyyy-MM-dd, and the offset rule to +1 day. To add more variables, click Add variable.

3. Add the OSS bucket for data storage and analysis.

   1. In the OSS Reference section, click Add OSS Reference.

   2. Select the destination OSS bucket.

      Note

      If you are not logged in to the OSS instance, enter your AccessKey ID and AccessKey secret in the Login instance dialog box.

   3. Specify the path in the OSS bucket to save the data.

      Note

      • If the path does not exist, it will be created automatically.

      • You can use variables in the path, such as /path/${foldername}.

   4. Enter oss as the reference alias for the OSS bucket in your Spark SQL statements.

   5. Click Save.

4. Add the MySQL database containing the order and product tables.

   1. In the Database reference section, click Add database reference.

   2. Select the database type, search for and select your database, and enter demo_id as the reference alias in your Spark SQL statements. For more information about the configuration items, see Database configuration table.

      Note

      If you are not logged in to the database, enter the database account and password in the Login instance dialog box.

   3. Click Save.

5. In the SQL editor, write your Spark SQL statements and click Save.

   bizdate is a system variable that requires no manual configuration.

   /* Write SQL statements by using Spark SQL syntax. Reference tables as alias.table_name. */
   /* Create a reference to the oss.t_order table and add dt as a partition field. */
   CREATE TABLE oss.t_order (
     id bigint COMMENT 'The primary key.',
     product_id bigint  COMMENT 'The product ID.',
     gmt_create timestamp  COMMENT 'The creation time.',
     gmt_modified timestamp  COMMENT 'The modification time.',
     customer_id bigint COMMENT 'The customer ID.',
     price decimal(38,8)  COMMENT 'The price.',
     status string COMMENT 'The order status.',
     province string COMMENT 'The transaction province.',
     dt string comment 'The business date partition.'
   )  partitioned by (dt) COMMENT 'The order table.';
   insert overwrite oss.t_order partition(dt='${bizdate}')
   select id, product_id, gmt_create, gmt_modified, customer_id, price, status, province 
   from demo_id.t_order o 
   where o.gmt_create>= '${bizdate}' and o.gmt_create< '${today}';
   /* Create a reference to the oss.t_product table. */
   CREATE TABLE oss.t_product (
     id bigint COMMENT 'The primary key.',
     name string COMMENT 'The product name.',
     type string COMMENT 'The product category.',
     gmt_create timestamp  COMMENT 'The creation time.',
     gmt_modified timestamp  COMMENT 'The modification time.'
   )   COMMENT 'The product table.';
   /* Fully synchronize the product table data. */
   insert overwrite oss.t_product 
   select id, name, type, gmt_create, gmt_modified 
   from demo_id.t_product; 
   /* Create a reference to the oss.t_order_report_daily table and use dt as a partition field. */
   CREATE TABLE oss.t_order_report_daily(
      dt string  comment 'The business date.',
      product_type string  comment 'The product category.',
      order_cnt bigint  comment 'The number of orders.',
      order_amt decimal(38, 8)  comment 'The order amount.'
   )  partitioned by (dt) comment 'The daily order statistics table.';
   /* Insert data by partition. */
   insert overwrite oss.t_order_report_daily partition(dt='${bizdate}')
   select
          p.type as product_type,
          count(*)  order_cnt,
          sum(price)  order_amt
     from oss.t_product p join oss.t_order o on o.product_id= p.id
    where o.gmt_create>= '${bizdate}'
      and o.gmt_create< '${today}'
    group by product_type;
                           

   OSS supports four file formats: CSV, Parquet, ORC, and JSON. The default format is CSV. You can specify a different format by using the USING clause in the CREATE TABLE statement.

   For example, to specify the Parquet format for the table:

   CREATE TABLE oss.t_order (
     id bigint COMMENT 'The primary key.',
     product_id bigint  COMMENT 'The product ID.',
     gmt_create timestamp  COMMENT 'The creation time.',
     gmt_modified timestamp  COMMENT 'The modification time.',
     customer_id bigint COMMENT 'The customer ID.',
     price decimal(38,8)  COMMENT 'The price.',
     status string COMMENT 'The order status.',
     province string COMMENT 'The transaction province.',
     dt string comment 'The business date partition.'
   )  USING PARQUET partitioned by (dt) COMMENT 'The order table.';

Run and publish the Spark SQL node

1. On the task flow details page, click Try run in the upper-left corner of the canvas.

   Click the Execution logs tab to view the results.

   • If the last line of the execution log shows status SUCCEEDED, the try run was successful.

   • If the last line of the execution log shows status FAILED, the try run failed.

     Note

     If the try run fails, check the execution log for the failed node and the cause. Then, modify the node configuration and try again.

   After the run succeeds, go to the DMS console homepage. In the instance list on the left, right-click your OSS bucket and click Query to view the synchronized and analyzed data.

2. Configure scheduling.

   1. Click a blank area of the canvas.

   2. Click the Task flow information tab.

   3. In the Scheduling Settings section, turn on the Enable Scheduling switch and configure the schedule. For detailed configuration information, see Scheduling cycle configuration table.

3. Publish the task flow. Its tasks will then run automatically based on the configured schedule.

   1. In the upper-left corner of the canvas, click Publish.

   2. Enter information in the Remarks field, and then click Publish.