Note:
Estimated duration: 60 minutes. Learning objective: Using an e-commerce order analysis scenario as an example, run through the complete data engineering pipeline end to end, from MySQL → data lakehouse → PySpark processing → workflow orchestration → scheduled scheduling → Ops monitoring.
Scenario Introduction
You are a data engineer who has just joined the data team. The business team wants to automatically sync order transactions from MySQL to the data lakehouse on a daily basis, and aggregate them by region to produce revenue summary tables for downstream BI and Ops analysis.
This tutorial will walk you through:
1. Register the MySQL data source in the workspace, and confirm that the compute resources required by the data lakehouse are ready.
2. Load the amazon_orders table from MySQL into the data lakehouse Catalog through offline sync.
3. Use a PySpark Notebook in Studio to complete data cleansing and aggregation by region, and write the results back to the data lakehouse.
4. Connect the offline sync and the Notebook into a DAG workflow.
5. Configure a scheduled schedule at 00:00 every day.
6. Track execution status and view run logs during workflow runs, and complete a failed rerun.
After completing this tutorial, you will master the full set of operations that data engineers perform most frequently in DataBuddy on a daily basis.
Prerequisites
Before starting this tutorial, make sure the following prerequisites are met:
You have activated the DataBuddy service and completed workspace initialization.
The workspace is bound to the following three types of compute resources. For more information, see Compute Resource Overview. Data ingestion compute resources (used for offline sync tasks).
Job cluster (used for Notebook tasks).
Interactive cluster (used for SQL debugging on demand).
An accessible MySQL instance is ready. The sample table is wedata_demo_db.amazon_orders, which contains 14 fields and about 20 rows of sample data. For more information, see the data preparation instructions in Step 1.
The workspace is bound to a data lakehouse Catalog (default: DataLakeCatalog), and the current account has table creation and read/write permissions on the target Schema wedata_demo_db.
Note:
The sample table wedata_demo_db.amazon_orders used in this tutorial is for demonstration purposes only. You can replace it with any structured data source, and the sync and PySpark steps remain applicable.
Sample Tutorial Data
Table fields of amazon_orders:
Field | Type | Description |
order_id
| VARCHAR(20) | Order number |
user_id
| VARCHAR(20) | User ID |
product_id
| VARCHAR(20) | Product ID |
product_name
| VARCHAR(200) | Product name |
category
| VARCHAR(50) | Product category |
unit_price
| DECIMAL(10,2) | Unit Price |
quantity
| INT | Quantity |
total_amount
| DECIMAL(12,2) | Order Amount |
order_status
| VARCHAR(20) | Order status: Delivered / Shipped / Cancelled, and so on. |
payment_method
| VARCHAR(30) | Payment method. |
order_date
| VARCHAR(10) | Order date (dd/MM/yyyy) |
state_code
| VARCHAR(5) | State code |
state
| VARCHAR(50) | State name |
city
| VARCHAR(100) | City |
Step 1: Registering a MySQL Data Source
Registering a data source is the foundation for all subsequent sync tasks. The data lakehouse Catalog is usually pre-bound by an administrator, so in this step you only need to provide the connection information for MySQL.
1. Log in to the DataBuddy console, click the workspace name in the workspace list to enter the workspace, and then choose Administration > Data Source Management in the left navigation pane. 2. In the upper-right corner of the page, click Create Data Source and select MySQL as the type.
3. Enter the connection information:
Data source name: for example, quickstart_mysql, which is used for reference in subsequent tasks.
Connection string: for example, jdbc:mysql://<host>:3306/wedata_demo_db.
Username and password: access credentials for the MySQL instance.
1. Select a Data Ingestion compute resource, click Test Connectivity, and after the status shows Connected, click Create and Ingest Data.
Note:
The data lakehouse Catalog (such as DataLakeCatalog) is automatically generated by the platform after the workspace is bound to a job cluster, so no manual creation is required. If the Catalog is not available, contact the workspace administrator to check the job cluster binding status. For details, see Add Data Source Connection.
Step 2: Synchronizing MySQL Tables to the Data Lakehouse
Next, create an offline sync task to load the entire amazon_orders table into the data lakehouse Catalog.
1. In the left navigation pane, select Data Integration, then select Ingestion Tasks, and click Create Task.
2. Enter the basic information:
Data source type: MySQL.
Ingestion Method: Offline Single-Table Ingestion.
Task name: for example, quickstart_mysql2dlc (it is recommended to add your name prefix to avoid conflicts among multiple users).
1. Configure the source:
Data source type: MySQL.
Data source connection: select quickstart_mysql created in Step 1.
Source database: wedata_demo_db.
Source data table: amazon_orders.
1. Configure the destination:
Target Catalog: DataLakeCatalog.
Target Schema: wedata_demo_db (if it does not exist, create it in advance or contact the administrator).
Target Table: for example, quickstart_amazon_orders. If the target table does not exist, click Create Target Table. The platform will generate the table creation SQL based on the source Schema. Manually adjust it if necessary and then submit.
Write mode: overwrite (recommended for initial full load; for incremental scenarios, use append or upsert instead).
1. Go to the field mapping step and click Map Fields with the Same Name to align source fields with target fields that have the same names one by one.
2. Click Save and then click Run to perform a debug run. After the status changes to Success, proceed to the next step.
3. Click Save and Publish. The task enters the published state. You can then select this task in the Step 4 workflow to orchestrate the workflow. Note:
For complete field descriptions, write mode differences, and advanced configuration references for offline sync tasks, see Table-to-Table Sync. Step 3: Developing and Debugging a PySpark Notebook in Studio
After the data lands in the data lakehouse, use PySpark to cleanse it, aggregate it by region, and write the aggregated results back to the data lakehouse.
3.1 Creating a Notebook and Configuring the Kernel
1. Select Development Workbench in the left sidebar.
2. Click + Create > Notebook. For example, use quickstart_transform.ipynb as the file name.
3. In the top right area, click "Not Connected", search for the relevant job cluster resource, and connect to it.
4. Wait until the Kernel status changes to the job cluster name. The status then changes to Connected.
Note:
For detailed Kernel configuration, kernel reuse, and runtime environment management, see Notebook Execution. 3.2 Preparing Sample Data (Optional)
If you have not completed the offline sync in Step 2, you can create a new SQL cell in the Notebook to directly create the target table and insert sample data.
-- Create a new SQL cell in the Notebook (select SQL from the language dropdown in the upper-left corner of the cell), and first execute the following table creation statement.
-- 1. Create the Schema if it does not exist.
CREATE SCHEMA IF NOT EXISTS `DataLakeCatalog`.`wedata_demo_db`;
-- 2. Create the table if it does not exist.
CREATE TABLE IF NOT EXISTS `DataLakeCatalog`.`wedata_demo_db`.`quickstart_amazon_orders` (
`'order_id' STRING COMMENT 'Order ID',
`'user_id' STRING COMMENT 'User ID',
`'product_id' STRING COMMENT 'Product ID',
`'product_name' STRING COMMENT 'Product name',
`}
`'unit_price' DECIMAL(10,2) COMMENT 'Unit price',
`'quantity' INT COMMENT 'Quantity',
`'total_amount' DECIMAL(12,2) COMMENT 'Order amount',
`order_status` STRING COMMENT 'Order status',
`payment_method` STRING COMMENT 'Payment method',
`'order_date' STRING COMMENT 'Order date',
`'state_code' STRING COMMENT 'State code',
`'state' STRING COMMENT 'State name',
`city` STRING COMMENT 'City'
) COMMENT 'Sample order table';
After the table is created, create another SQL cell to insert data:
-- 3. Insert sample data.
INSERT INTO `DataLakeCatalog`.`wedata_demo_db`.`quickstart_amazon_orders`
(`order_id`, `user_id`, `product_id`, `product_name`, `category`, `unit_price`, `quantity`, `total_amount`, `order_status`, `payment_method`, `order_date`, `state_code`, `state`, `city`)
VALUES
('ORD-001','U001','P001','Towel Set','Home',CAST(15.99 AS DECIMAL(10,2)),2,CAST(31.98 AS DECIMAL(12,2)),'Delivered','CreditCard','01/03/2024','CA','California','Los Angeles'),
('ORD-002','U002','P002','Bedsheet','Home',CAST(25.50 AS DECIMAL(10,2)),1,CAST(25.50 AS DECIMAL(12,2)),'Shipped','PayPal','02/03/2024','NY','New York','New York City'),
('ORD-003','U003','P003','Curtain','Home',CAST(40.00 AS DECIMAL(10,2)),3,CAST(120.00 AS DECIMAL(12,2)),'Delivered','CreditCard','03/03/2024','TX','Texas','Houston'),
('ORD-004','U004','P004','Carpet','Home',CAST(120.50 AS DECIMAL(10,2)),1,CAST(120.50 AS DECIMAL(12,2)),'Cancelled','ApplePay','04/03/2024','FL','Florida','Miami'),
('ORD-005','U005','P005','Pillow','Home',CAST(12.99 AS DECIMAL(10,2)),4,CAST(51.96 AS DECIMAL(12,2)),'Delivered','CreditCard','05/03/2024','WA','Washington','Seattle'),
('ORD-006','U006','P006','Mattress','Home',CAST(299.99 AS DECIMAL(10,2)),1,CAST(299.99 AS DECIMAL(12,2)),'Delivered','DebitCard','06/03/2024','IL','Illinois','Chicago'),
('ORD-007','U007','P007','Blanket','Home',CAST(35.00 AS DECIMAL(10,2)),2,CAST(70.00 AS DECIMAL(12,2)),'Shipped','CreditCard','07/03/2024','AZ','Arizona','Phoenix'),
('ORD-008','U001','P001','Towel Set','Home',CAST(15.99 AS DECIMAL(10,2)),5,CAST(79.95 AS DECIMAL(12,2)),'Delivered','CreditCard','08/03/2024','CA','California','San Diego'),
('ORD-009','U008','P008','Rug','Home',CAST(65.00 AS DECIMAL(10,2)),1,CAST(65.00 AS DECIMAL(12,2)),'Delivered','PayPal','09/03/2024','NV','Nevada','Las Vegas'),
('ORD-010','U009','P009','Comforter','Home',CAST(85.00 AS DECIMAL(10,2)),1,CAST(85.00 AS DECIMAL(12,2)),'Shipped','CreditCard','10/03/2024','OR','Oregon','Portland');
After successful execution, you can run the following query to verify the data:
-- 4. Verify that the data has been inserted.
SELECT * FROM `DataLakeCatalog`.`wedata_demo_db`.`quickstart_amazon_orders`;
After successful execution, switch back to the PySpark cell to proceed to the next step.
3.3 Reading Data Lakehouse Tables
In the first PySpark cell, read the table synchronized in Step 2 (or the table manually created in Step 3.2):
df_orders = spark.sql("""
SELECT *
FROM `DataLakeCatalog`.`wedata_demo_db`.`quickstart_amazon_orders`
""")
display(df_orders)
print(f"Total rows: {df_orders.count()}")
Tables in a data lakehouse are referenced by using the three-part name Catalog.Schema.Table.
3.4 Cleansing and Dimension Derivation
from pyspark.sql import functions as F
# Filter canceled orders
df_valid = df_orders.filter(F.col("order_status") != "Cancelled")
# Parse the dd/MM/yyyy text date into a date type and derive the year and month
df_valid = (
df_valid
.withColumn("order_date_parsed", F.to_date("order_date", "dd/MM/yyyy"))
.withColumn("order_year", F.year("order_date_parsed"))
.withColumn("order_month", F.month("order_date_parsed"))
)
display(df_valid.select("order_id", "order_date", "order_date_parsed", "order_year", "order_month"))
3.5 Aggregating by State Dimension
df_state_revenue = (
df_valid.groupBy("state", "state_code")
.agg(
F.count("order_id").alias("order_count"),
F.sum("total_amount").alias("total_revenue"),
F.avg("total_amount").alias("avg_order_value"),
F.countDistinct("user_id").alias("unique_customers"),
)
.orderBy(F.desc("total_revenue"))
)
display(df_state_revenue)
3.6 Creating a Target Table and Writing Back to the Data Lakehouse
spark.sql("""
CREATE TABLE IF NOT EXISTS `DataLakeCatalog`.`wedata_demo_db`.`quickstart_state_revenue` (
state STRING COMMENT 'State name',
state_code STRING COMMENT 'State code',
order_count BIGINT COMMENT 'Order count',
total_revenue DOUBLE COMMENT 'Total revenue',
avg_order_value DOUBLE COMMENT 'Average order value',
unique_customers BIGINT COMMENT 'Unique customers',
) COMMENT 'Revenue summary by state'
""")
(
df_state_revenue
.select("state", "state_code", "order_count", "total_revenue", "avg_order_value", "unique_customers")
.write.mode("overwrite")
.insertInto("`DataLakeCatalog`.`wedata_demo_db`.`quickstart_state_revenue`")
)
display(spark.sql("SELECT * FROM `DataLakeCatalog`.`wedata_demo_db`.`quickstart_state_revenue`"))
After confirming that the output table contains data, click Save at the top of the editor to generate a version snapshot.
Note:
Studio enables auto-saving drafts by default, but only manually saved versions can be referenced by workflow tasks. Remember to click Save after each modification to the Notebook. For more information, see Notebook IDE Basic Operations. Step 4: Orchestrating Tasks into a Workflow
A Workflow is the smallest unit of unified scheduling. In this step, an offline sync task and a Notebook are connected into a DAG.
4.1 Creating a Workflow
1. In the left navigation pane, choose Data Engineering > Workflow.
2. Click Create Workflow to automatically enter the workflow editor canvas.
4.2 Adding Two Tasks to the Canvas
Add Offline Data Ingestion Task:
1. On the canvas, click + Add Task and select Offline Data Sync as the task type.
2. Task Name: sync_mysql2dlc.
3. Task Source: select quickstart_mysql2dlc created in Step 2 from the drop-down list.
4. Compute Resources: Select a data ingestion resource group.
5. Click Create Task.
Add Notebook Task:
1. Click + Add Task again and select Notebook as the task type.
2. Task Name: pyspark_transform.
3. Task Source: Workspace.
4. Path: Click the input box and select quickstart_transform.ipynb created in Step 3.
5. Compute Resources: Select the job cluster used for Notebook debugging. Keep the resource mode and resource configuration at their default values.
6. Click Create Task.
4.3 Connecting Dependencies
In the canvas, move the mouse to the lower anchor point of the sync_mysql2dlc node, hold down the left button and drag to the upper anchor point of the pyspark_transform node, then release to create a dependency arrow.
The final DAG looks like this:
sync_mysql2dlc ───────▶ pyspark_transform
(Offline Data Ingestion) (Notebook)
Click the pyspark_transform node and confirm that sync_mysql2dlc is listed in Upstream Dependencies on the right, with Run Condition remaining at the default All Successful.
4.4 Debugging and Running the Entire Workflow
1. Click Run at the top of the canvas to immediately trigger a debug run.
2. Observe the execution order of tasks in the workflow: sync_mysql2dlc runs first, and after it succeeds, pyspark_transform starts automatically.
3. After clicking a run record, click the task node to view details in Run Logs and Data Reconciliation: number of rows, Spark Job status, error messages, and more.
If both tasks show Success, the DAG link is connected.
Note:
For general rules on task configuration items (task name, upstream dependencies, run conditions, parameters, monitoring metrics, alarms, and failure and timeout policies), see Task General Configuration. Step 5: Configuring Scheduled Triggering
After debugging passes, configure the workflow to run automatically every day.
1. In the right sidebar of the workflow editor, click Schedule Configuration > Add Schedule.
2. Fill in the scheduling parameters:
Scheduling Status: Started.
Trigger Method: Scheduled trigger.
Scheduling Time Zone: Select based on your business needs (for example, (UTC+08:00) Beijing, Chongqing, Hong Kong SAR, Urumqi).
Scheduling Time: Set the start time to today, and use the default 2099-12-31 as the end time.
Configuration Method: Regular.
Scheduling Cycle: Day.
Execution Time: Single point in time. Enter 00:00.
1. Confirm that the next five estimated run times displayed in the Schedule Preview area meet your expectations, including the date and time for the first, second, ... and fifth run.
2. Click OK to save the scheduling configuration.
Note:
For more complex scheduling scenarios, such as hourly scheduling, weekly scheduling on specified days, Cron expressions, file arrival triggers, and continuous running, refer to Scheduling and Triggers. After you save the workflow, it automatically generates periodic instances according to the schedule. You can return to the workflow editor at any time to modify tasks or schedules. Click Save to apply the changes, and the latest code and configuration will be used in the next scheduling cycle.
Step 6: Monitoring and Troubleshooting Workflow Runs
After you configure scheduled scheduling, daily Ops is mainly performed in the Workflow Run module.
6.1 Viewing the Run List
1. In the left navigation pane, choose Workflow > Workflow Run.
2. At the top, filter by Workflow Name for quickstart_orders_etl to view all run instances of this workflow and the status trend chart.
3. Click a Run Stream ID to go to the run details.
6.2 Locating Issues on the Run Details Page
The run details page provides three views:
View | Scenario |
Graph mode (DAG) | View the status of each task at a glance. Node colors: green = success / red = failure / blue = running / gray = waiting / yellow = failed retry. |
List mode | Display the run ID, type, duration, compute resources, and source file path of all tasks in a table. |
Gantt chart | Display the start and end time range of each task along a timeline to help identify long-running tasks and bottlenecks. |
In the DAG, click any task node to go to the Task Run Details page:
Run Results: Displays the task code and outputs.
Run Logs: Error codes, error messages, and complete logs, with support for search, download, and auto-refresh.
Task Insights: Performance metrics such as cumulative CPU time, scanned data volume, Shuffle size, and the number of output rows.
Spark UI in the upper-right corner: Click the button to go to the native Spark UI for troubleshooting task details.
6.3 Rerunning Failed Tasks
When a task fails due to data latency, network jitter, or temporary configuration issues:
1. On the workflow run details page, identify the failed tasks and their root causes by referring to the run logs.
2. Fix the root cause (for example, modify the Notebook code and then click Save again in Studio).
3. On the workflow run details page, click Rerun. In the DAG dialog box, failed nodes are selected by default, and you can choose whether to include their upstream nodes.
4. Click Run to trigger a rerun. The rerun uses the latest task code and configuration.
Note:
If alarms are configured for the task, the alarm channel will still be triggered during a rerun. To mute alarms during troubleshooting reruns, enable the do-not-disturb option in the alarm configuration in advance.
Step 7: Cleaning Up Resources (Optional)
If you no longer need the resources created in this tutorial, clean them up in the following order to avoid ongoing billing and quota usage:
1. Pause Schedule: In the workflow editor, click Pause Schedule under Schedule Configuration to stop subsequent scheduled instances.
2. Delete Run Records: Filter this workflow in the workflow run list, and delete the run records that have reached a terminal state.
3. Delete Workflow: Click Delete at the end of the workflow row in the workflow list. You must first terminate all non-terminal runs.
4. Delete Target Table (Optional): Execute SQL in Studio, or delete it in the Catalog data directory:
DataLakeCatalog.wedata_demo_db.quickstart_amazon_orders
DataLakeCatalog.wedata_demo_db.quickstart_state_revenue
1. Delete Data Source: Delete quickstart_mysql in Platform Management > Data Source Management if it is no longer referenced by other tasks.
Warning:
Workflows and run records cannot be recovered after deletion. Before you proceed, make sure there are no business dependencies.
Verification Result
After completing all steps, you should confirm:
1. The offline sync task quickstart_mysql2dlc has been executed successfully. The target table DataLakeCatalog.wedata_demo_db.quickstart_amazon_orders contains data, and the number of rows is consistent with the MySQL source.
2. The Notebook file quickstart_transform.ipynb has been saved as a version in Studio, and its cells can be run independently and successfully.
3. The summary target table DataLakeCatalog.wedata_demo_db.quickstart_state_revenue presents metrics such as number of orders, total revenue, average order value, and number of unique users by state.
4. The workflow quickstart_orders_etl is visible in the workflow list, with its scheduling status set to Started. Both tasks in the DAG are shown as successful.
5. There is at least one run instance with the Succeeded status in Workflow Runs, and the task run logs, insights, and Spark UI are all accessible.
Summary
In this tutorial, you have completed:
Register a MySQL data source and test its connectivity in Platform Management > Data Source Management.
Create and run an offline sync task (MySQL → data lakehouse) in Data Integration > Task Management.
Use a PySpark Notebook in Studio to perform data cleansing and regional aggregation, and write the results back to the data lakehouse.
In Workflow, connect the offline data integration task and the Notebook task into a DAG, and configure the upstream and downstream dependencies.
Configure a daily schedule at 00:00.
In Workflow Runs, track the run status, view logs, identify failed tasks, and complete a rerun.
This end-to-end pipeline covers more than 80% of the high-frequency operations that data engineers perform daily. You can then expand it with more task types (SQL, Python File, Ray, For Each, IF ELSE, and so on) and advanced capabilities (parameter passing, alarms, quality monitoring, and CI/CD) based on business complexity.