Configure a cross-database Spark SQL node

Limits

• Spark SQL task nodes run on the Spark compute engine. Do not process more than 2 million data records in a single task at a time. Otherwise, task execution efficiency may be affected.

• Due to limited computing resources, the timeliness of computing cannot be guaranteed during peak hours.

• Processing a large table that does not have a primary key can cause a memory overflow or an out-of-memory (OOM) error.

• If a recurring schedule node runs successfully but then fails 10 or more consecutive times, the batch integration job fails immediately and no more Spark tasks are submitted. In this case, you must manually run the task node until it succeeds.

Scenarios

Cross-database Spark SQL tasks are mainly used for cross-database data synchronization and transformation:

• Cross-database data synchronization:

  • Synchronize data from an online database to a data warehouse for data transformation. Online services generate large amounts of data. To process and analyze this data, you typically need to synchronize it to a data warehouse designed for data transformation and analytics.

  • Synchronize data from a data warehouse back to an online database for data queries. After data is processed and analyzed in a data warehouse, it often needs to be synchronized back to the online database. This allows online applications to provide related data analytics and statistical services.

  Example: A consumer service platform uses a MySQL database. You need to perform statistical analysis on consumption data, such as transaction amounts and the number of transactions, in an AnalyticDB for PostgreSQL data warehouse. The analyzed data is then sent back to the consumer service platform for users to query online.

  • To synchronize incremental consumption data from the MySQL database to the AnalyticDB for PostgreSQL instance, use the following Spark SQL statement:

    INSERT INTO adb_dw.orders
    SELECT * FROM mysql_db.orders
    WHERE dt>${bizdate} AND dt<=${tomorrow}

  • To synchronize the processed data from the AnalyticDB for PostgreSQL instance to the MySQL database, use the following Spark SQL statement:

    INSERT INTO mysql_db.orders_month 
    SELECT * FROM adb_dw.orders_month 
    WHERE dt=${bizdate}

• Cross-database data transformation:

  You can write data from multiple databases to an online database. This lets you query the data directly in your online application.

  Example: An e-commerce company's transaction data is stored in an online MySQL database. Its employee data is stored in an HR system that uses an Oracle database. When the company needs to calculate sales by department, it must perform a join query on department, employee, and transaction data. The following Spark SQL statement joins the `sales` table from the `mysql_db` online database with the `users` table from the `oracle_db` database. It then groups the joined data by department name to count the number of transactions and sum the transaction amounts. The processed data is written to the `mysql_db` online database.

  INSERT INTO mysql_db.dept_sales 
  SELECT dept_name, trade_date, COUNT(*) cnt, SUM(amt) amt FROM mysql_db.sales t1 JOIN oracle_db.users t2 ON t1.emp_id=t2.id
  WHERE t1.trade_date=${bizdate} GROUP BY t2.dept_name

Features

• Cross-database data processing: You can use SQL statements to operate on data in different databases. The data ecosystem is comprehensive and can be extended to support a variety of data sources.

• Large data volume processing: Supports the fast processing of large-scale data (over 100,000 records).

• Spark SQL syntax: Deployed based on Spark 3.1.2, it provides all syntax features and native functions of this version. Native functions include aggregate functions, window functions, array functions, map functions, date and time processing functions, and JSON processing functions.

• Standard SQL compatibility: You can use standard SQL statements to perform cross-database data synchronization and transformation.

• Serverless: Spark SQL tasks are a serverless computing service that uses the Spark engine for data processing. You do not need to purchase or maintain computing resources. There are no O&M or upgrade costs.

• Supported SQL statements include the following: CREATE TABLE, CREATE SELECT, DROP TABLE, INSERT, INSERT SELECT, ALTER TABLE, TRUNCATE, SET, ANALYZE, and MSCK REPAIR.

  Note

  • The TRUNCATE statement supports only OSS tables. It has no effect on other table types.

  • Unsupported SQL statements include the following: SELECT, DELETE, UPDATE, CREATE DATABASE, and DROP DATABASE.

• Supports OSS file storage data sources, including CSV, JSON, PARQUET, and ORC file formats.

Procedure

1. Log on to the DMS console V5.0.

2. In the top navigation bar, choose Data+AI > Data+AI > Task Orchestration.

   Note

   If you use the DMS console in simple mode, move the pointer over the icon in the upper-left corner and choose All Features > Data+AI > Data Development > Task Orchestration.

3. Click the name of the target task flow to go to its details page.

   Note

   For instructions on adding a task stream, see Add a task stream.

4. In the Task Type list on the left of the canvas, drag the Cross-database Spark SQL node to the blank area of the canvas.

5. Double-click the Cross-database Spark SQL node.

6. Optional: On the Cross-database Spark SQL configuration page, click Variable Settings to configure variables to reference in your SQL statements. You can click the icon in the upper-right corner of the Variable Settings area to view tips on variable configuration.

   • Click the Node Variable tab to configure node variables. For more information, see Configure time variables.

   • Click the Task Flow Variable tab to configure task flow variables. For more information, see Configure time variables.

   • Click the Input Variables tab to view upstream variables, runtime status variables, and system variables.

7. Optional: In the OSS Reference area, click Add OSS Reference to add an OSS bucket to reference in your SQL statements. After configuration, click Save.

   Configuration Item	Required	Description
   Database	Yes	Search for and select the target database from the drop-down list. Important For a database not in Security Collaboration mode, if the database account changes (such as password or permission changes), the task may fail. You must log on to the database again and save the node configuration.
   OSS Path	Yes	Specify the path where data is stored in the OSS Bucket. Note If the path does not exist, it is automatically created. The path supports variables, such as /path/${foldername}.
   Spark SQL reference alias	Yes	Enter the alias for the OSS bucket in the Spark SQL statement. The default value is oss. Note The reference alias can contain digits, uppercase letters, lowercase letters, and underscores (_). It can be up to 32 characters long.

8. Optional: In the Database Reference area, click Add Database Reference to add a database to reference in your SQL statements. After configuration, click Save.

   Configuration Item	Required	Description
   Database Type	Yes	Select the type of the target database. Note Go to the SQL window. In the instance list on the left, move the mouse pointer over the instance to which the database belongs to view the database type.
   Database	Yes	Search for and select the target database from the drop-down list. Important For a database not in Security Collaboration mode, if the database account changes (such as password or permission changes), the task may fail. You must log on to the database again and save the node configuration.
   Spark SQL reference alias	Yes	Enter the alias for the database in the Spark SQL statement. The default value is the database name. Note The database alias can contain digits, uppercase letters, lowercase letters, and underscores (_). It can be up to 32 characters long.

   To add multiple target databases, click the icon to the right of the database.

9. In the SQL area, write Spark SQL statements and perform a test run.

   1. Write the Spark SQL statements and click Save.

      For example, to read data from the `ex_customer` table in the `qntext` database and write it to the `test_table` table in the `qn_rds` database, set the alias for the `qntext` database to `qn` and the alias for the `qn_rds` database to `rds`. The SQL statement is as follows:

      INSERT INTO rds.test_table SELECT * FROM qn.ex_customer;

      Note

      • You can reference variables in SQL statements using the `${var_name}` format.

      • While writing SQL code, you can click SQL Preview at any time to preview the code.

      • Click Validity Check to check the validity of the SQL code.

   2. Click Test Run.

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

      • If the last line of the execution log shows status FAILED, the test run failed. View the execution log to identify the reason for the failure, modify the configuration, and try again.

Related documents

• Use task orchestration to synchronize data across databases

• Use a cross-database Spark SQL task to synchronize data to OSS