MaxCompute:Use PyODPS in DataWorks

Memory

If memory usage exceeds the node limit, the node reports Got killed and terminates. To avoid this, push data processing tasks to MaxCompute for distributed execution instead of downloading data to the DataWorks environment and processing it locally. For a comparison of the two approaches, see the "Precautions" section in Overview.

Data records

By default, options.tunnel.use_instance_tunnel is False in DataWorks PyODPS nodes. This means instance.open_reader uses the Result interface, which returns up to 10,000 data records and has limited support for complex data types. To read all records or handle complex types such as Arrays, enable InstanceTunnel. See Read SQL execution results for details.

Packages

Most common data science packages come pre-installed (see Pre-installed packages). The following restrictions apply:

• atexit is not supported. Use try-finally instead.

• matplotlib is not available, which affects the DataFrame plot function.

• DataFrame UDFs run in a Python sandbox and can only use pure Python libraries and NumPy. Other third-party libraries such as pandas are not supported inside UDFs. For non-UDF operations, NumPy and pandas are available.

• Third-party packages that contain binary code are not supported.

Read SQL execution results

Because InstanceTunnel is not enabled by default in DataWorks, instance.open_reader is limited to 10,000 records and may not support complex data types. If your project does not have data protection enabled and you need to read all records or use complex types such as Arrays, enable InstanceTunnel.

Option 1: Enable globally

Applies to all subsequent open_reader calls in the node.

options.tunnel.use_instance_tunnel = True
options.tunnel.limit_instance_tunnel = False  # Remove the 10,000-record limit.

with instance.open_reader() as reader:
    # InstanceTunnel is active. Use reader.count to get the total number of records.
    pass

Option 2: Enable per reader

Applies only to the current open_reader call.

with instance.open_reader(tunnel=True, limit=False) as reader:
    # InstanceTunnel is active for this reader. All records are accessible.
    pass

For more information, see Obtain the execution results of SQL statements.

Pre-installed packages

The following packages are pre-installed in DataWorks nodes:

Install a custom package

If the package you need is not pre-installed, use pyodps-pack to bundle it and load_resource_package to load it in the node.

1. Bundle the package. The following example bundles the ipaddress package:

   pyodps-pack -o ipaddress-bundle.tar.gz ipaddress

   For Python 2 nodes, add --dwpy27:

   pyodps-pack --dwpy27 -o ipaddress-bundle.tar.gz ipaddress

   To reduce bundle size, exclude packages that are already pre-installed in DataWorks:

   pyodps-pack -o bundle.tar.gz --exclude numpy --exclude pandas <your-package>

   The total size of downloaded packages cannot exceed 100 MB.

2. Upload and submit the .tar.gz file as a MaxCompute resource.

3. In the PyODPS node, load and import the package:

   load_resource_package("ipaddress-bundle.tar.gz")
   import ipaddress

For more information, see Generate a third-party package for PyODPS and Reference a third-party package in a PyODPS node.

Procedure

1. Grant the new account the required permissions on the project. See Appendix: Grant permissions to another account.

2. In the PyODPS node, create an entry object for the new account:

   import os
   
   # Store credentials in environment variables rather than hardcoding them.
   new_odps = o.as_account(
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
   )

3. Verify that the account switch succeeded by checking the current user:

   print(new_odps.get_project().current_user)

   If the output matches the AccessKey ID of the new account, the switch was successful.

Example

This example creates a table, queries it using a different account, and prints the results.

1. Create the pyodps_iris table and import sample data. For instructions, see Create tables and upload data.

   CREATE TABLE IF NOT EXISTS pyodps_iris
   (
     sepallength  DOUBLE COMMENT 'sepal length (cm)',
     sepalwidth   DOUBLE COMMENT 'sepal width (cm)',
     petallength  DOUBLE COMMENT 'petal length (cm)',
     petalwidth   DOUBLE COMMENT 'petal width (cm)',
     name         STRING COMMENT 'type'
   );

2. Grant the new account permissions on the project and table. See Appendix: Grant permissions to another account.

3. Create a PyODPS 3 node and run the following code. For instructions, see Develop a PyODPS 3 task.

   from odps import ODPS
   import os
   
   # Store credentials in environment variables rather than hardcoding them.
   os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'] = '<your-access-key-id>'
   os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'] = '<your-access-key-secret>'
   
   od = o.as_account(
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
   )
   
   # Query rows where sepallength > 5.
   with od.execute_sql('SELECT * FROM pyodps_iris WHERE sepallength > 5').open_reader() as reader:
       print(reader.raw)
       for record in reader:
           print(record["sepallength"], record["sepalwidth"],
                 record["petallength"], record["petalwidth"], record["name"])
   
   # Verify the current user.
   print(od.get_project().current_user)

4. Run the node. The output looks similar to the following:

   Executing user script with PyODPS 0.11.4.post0
   
   "sepallength","sepalwidth","petallength","petalwidth","name"
   5.4,3.9,1.7,0.4,"Iris-setosa"
   ...
   <User 139xxxxxxxxxxxxx>