DataWorks:Document

Create and publish a Python resource

1. In the left-side navigation pane of DataWorks DataStudio, click to go to Resource Management.

2. Right-click the target directory or click the + icon in the upper-right corner. Select Create Resource > MaxCompute Python, and name the file my_utils.py.

3. In the File Content section, click Edit Online, paste your utility function code, and click Save.

   # my_utils.py
   def greet(name):
       return f"Hello, {name} from resource file!"

4. Click Save, then Publish in the toolbar. The resource becomes available to both development and production tasks.

Reference the resource in a Notebook

In the first line of a Python cell, use ##@resource_reference to reference the published resource:

##@resource_reference{"my_utils.py"}
# If the resource is in a subdirectory (e.g., my_folder/my_utils.py),
# reference it without the directory name: ##@resource_reference{"my_utils.py"}
from my_utils import greet

message = greet('DataWorks')
print(message)

Debug in the development environment

Run the Python cell. The output is:

Hello, DataWorks from resource file!

Publish to the production environment

After saving and publishing the Notebook node, go to Operation Center > Recurring Tasks and click Test. After the task succeeds, the output Hello, DataWorks from resource file! appears in the logs.

For more information, see MaxCompute resources and functions.

Debug in the development environment

1. Mount a dataset: On your personal development environment's details page, configure the dataset under Storage Configuration > Dataset.

2. Access in code: The dataset is mounted to a path in the personal development environment. Read from or write to this path directly:

   # Assume the dataset is mounted to /mnt/data/dataset.
   import pandas as pd
   
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # Write data to MaxCompute using PyODPS.
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"Data successfully written to MaxCompute table mc_test_table.")

Deploy to production

1. Mount a dataset: On the Notebook node editing page, add the dataset under scheduling configurations > scheduling policy in the right-side navigation pane.

2. Access in code: After committing and publishing the node, the dataset is mounted to a path in the production environment. Use the same path in your code:

   # Assume the dataset is mounted to /mnt/data/dataset.
   import pandas as pd
   
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # Write data to MaxCompute using PyODPS.
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"Data successfully written to MaxCompute table mc_test_table.")

For more information, see Use datasets in a personal development environment.

%odps — Get a PyODPS entry object

Returns an authenticated PyODPS object bound to the current MaxCompute project. This avoids hard-coding AccessKeys in your code.

o = %odps

After running the command, a compute resource selector appears in the lower-right corner with a project automatically selected. Click the project name to switch projects.

Use the object to run PyODPS scripts. For example, to list all tables in the current project:

with o.execute_sql('show tables').open_reader() as reader:
    print(reader.raw)

%maxframe — Establish a MaxFrame connection

Creates a MaxFrame session for distributed, pandas-like data processing on MaxCompute data.

# Connect to MaxCompute using MaxFrame.
mf_session = %maxframe

df = mf_session.read_odps_table('your_mc_table')
print(df.head())

# Destroy the session manually to release resources after debugging.
mf_session.destroy()

Engine comparison

After connecting to a Spark engine, the execution context of the entire Notebook kernel switches to the remote PySpark environment. Write PySpark code directly in subsequent cells.

MaxCompute Spark

Bind a MaxCompute compute resource before establishing a connection.

Connect through Livy to the Spark engine built into your MaxCompute project.

1. Establish a connection: Run the following command in a Python cell. The system automatically creates or reuses a Spark session.

   # Create a Spark session and start Livy.
   %maxcompute_spark

2. Execute PySpark code: Use the %%spark cell magic in a new Python cell.

   # Python cells using MaxCompute Spark must start with %%spark.
   %%spark
   
   df = spark.sql("SELECT * FROM your_mc_table LIMIT 10")
   df.show()

3. Release the connection: After debugging, stop or delete the session. In a production environment, the system automatically stops and deletes the Livy session when the task instance ends.

   # Stop the Spark session and Livy.
   %maxcompute_spark stop
   
   # Stop Livy and delete it.
   %maxcompute_spark delete

EMR Serverless Spark

Bind an EMR Serverless Spark compute resource to your workspace and create a Livy Gateway before establishing a connection.

Interact with EMR Serverless Spark by connecting to a Livy Gateway created in advance.

1. Establish a connection: Select the EMR compute resource and Livy gateway in the lower-right corner of the cell, then run one of the following commands:

   • Selected: global configuration overrides the session's custom parameters.

   • Not selected: the session's custom parameters override the global configuration.

   # Basic connection
   %emr_serverless_spark

   To pass custom Spark parameters, use %%emr_serverless_spark (two percent signs):

   %%emr_serverless_spark
   {
     "spark_conf": {
       "spark.emr.serverless.environmentId": "<EMR_Serverless_Spark_Runtime_Environment_ID>",
       "spark.emr.serverless.network.service.name": "<EMR_Serverless_Spark_Network_Connection_ID>",
       "spark.driver.cores": "1",
       "spark.driver.memory": "8g",
       "spark.executor.cores": "1",
       "spark.executor.memory": "2g",
       "spark.driver.maxResultSize": "32g"
     }
   }

   Custom parameters apply only to the current session. If omitted, the system uses the global parameters configured in Admin Center. To share configurations across tasks and users, set them globally in Admin Center > Serverless Spark > SPARK parameters. The Prioritize Global Configurations option in Admin Center controls priority when the same parameter appears in both places:

2. (Optional) Reconnect: If an administrator deletes the token on the Livy gateway page, recreate it with:

   # Reconnect and refresh the Livy token.
   %emr_serverless_spark refresh_token

3. Execute PySpark or SQL code: After a successful connection, the kernel switches. Write PySpark code directly in a Python cell, or write SQL in an EMR Spark SQL cell.

   • Submit SQL to EMR Serverless Spark using an EMR Spark SQL cell — the cell reuses the connection from %emr_serverless_spark and submits the job automatically. No compute resource selection is needed in the cell.

   • Submit PySpark code in a Python cell — no %%spark prefix is required.

4. Release the connection:

   Important

   If multiple users share a Livy Gateway, stop or delete affects all users currently on that gateway. Use these commands with caution.

   # Stop the Spark session and Livy.
   %emr_serverless_spark stop
   
   # Stop Livy and delete it.
   %emr_serverless_spark delete

AnalyticDB for Spark

Bind an AnalyticDB for Spark compute resource to your workspace before establishing a connection.

Connect to an AnalyticDB for Spark engine by creating a Spark Connect Server.

1. Establish a connection: Select an ADB Spark compute resource in the lower-right corner of the cell. Configure the vSwitch ID and Security Group ID to ensure network connectivity, then run:

   • vSwitch ID (`vswitchId`): In the Alibaba Cloud AnalyticDB MySQL console, view the vSwitch ID in Network Information on the instance details page.

   • Security Group ID (`securityGroupId`): In Network Settings on your personal development environment's details page, find the ID of the selected Security Group — it starts with sg-.

   Important

   To ensure network connectivity, select the same Virtual Private Cloud (VPC) and vSwitch as your AnalyticDB for Spark instance when creating the personal development environment.

   # Configure the vSwitch ID and Security Group ID for network connectivity.
   %adb_spark add  --spark-conf spark.adb.version=3.5  --spark-conf spark.adb.eni.enabled=true  --spark-conf spark.adb.eni.vswitchId=<vSwitch_ID_of_ADB>  --spark-conf spark.adb.eni.securityGroupId=<Security_Group_ID_of_personal_development_environment>

   How to find the vSwitch and Security Group IDs:

2. Execute PySpark code: After the connection is established, run PySpark in a new Python cell.

   The AnalyticDB for Spark engine can only process C-Store tables that have the 'storagePolicy'='COLD' attribute.

   # AnalyticDB for Spark can only process C-Store tables.
   df = spark.sql("SELECT * FROM my_adb_cstore_table LIMIT 10")
   df.show()

3. Release the connection: After debugging, clean up the connection session to save resources. In production, the system cleans up automatically.

   %adb_spark cleanup

Establish a connection

Run %lindorm_ray in a Python cell. A compute resource selector appears in the lower-right corner — select your Lindorm compute resource and the created RAY resource group.

# Connect to the specified Lindorm Ray resource group.
%lindorm_ray

Execute Ray code

After a successful connection, write and execute Ray code in a new Python cell. Logs stream back to the cell's output area in real time.

The following example defines a remote task using the @ray.remote decorator. The task runs on the Ray cluster, and its logs and result are returned to the output area:

import ray
import time

@ray.remote
def hello_world():
  print("Hello from Lindorm Ray!")
  time.sleep(5)
  return "Task finished."

# Submit the remote task.
result_ref = hello_world.remote()
print(ray.get(result_ref))

Specify custom startup parameters (optional)

To install third-party Python packages or upload local code files, use %%lindorm_ray to establish the connection with a custom runtime environment configuration.

Example 1: Install dependencies

Install the jieba package in the Ray environment using the pip parameter:

%%lindorm_ray
{
  "runtime_env": {
    "pip": ["jieba"]
  }
}

After the environment is ready, import and use the package in subsequent Ray tasks:

import ray

@ray.remote
def do_work(x):
    import jieba
    return "/".join(jieba.cut(x))

print(ray.get(do_work.remote("Welcome to the DataWorks+LindormRay solution")))

Example 2: Upload and use a DataWorks resource

Use the working_dir parameter to upload a resource from DataWorks Resource Management to the Ray cluster:

# Reference a resource from DataStudio Resource Management and declare its path.
%%lindorm_ray
{
    "runtime_env": {
        "working_dir": "/mnt/workspace/_dataworks/resource_references"
    }
}

Assume ray_resource.py has been uploaded to DataStudio Resource Management:

ray_resource.py:

def fun():
    print("This is a test function in ray_resource.py")

Reference and use it in a Ray task:

import ray

##@resource_reference{"ray_resource.py"}

@ray.remote
def do_work(x):
    print('Ray says:', x)
    from ray_resource import fun
    fun()
    return x

worker = do_work.remote("Welcome to the DataWorks+LindormRay solution")
print(ray.get(worker))

Production scheduling and O&M

After development and debugging, commit and publish the Notebook node. It runs as a Lindorm Ray node in a DAG on a periodic schedule.

• Parameterization: Use standard DataWorks scheduling parameters such as ${bizdate}.

• Log viewing: To prevent excessive logs from affecting performance, the system loads only the first 1 MB of logs by default. If logs are truncated, the output includes a link to the Lindorm console for the complete task logs.

• Resource release: After a scheduled production task ends, the Lindorm Ray task enters its desired state and releases resources. During interactive development, restart the kernel or close the Notebook to release resources.