MaxCompute:Develop a PyODPS 2 task

Use the MaxCompute entry point

The global variable odps (or o) is available in every PyODPS node. No client initialization is required.

print(odps.exist_table('PyODPS_iris'))

Execute SQL statements

Call execute_sql to run SQL. For full SQL support, see SQL.

Reading results: the 10,000-record default limit

By default, InstanceTunnel is disabled in DataWorks. When disabled, instance.open_reader uses the Result interface, which returns at most 10,000 records. Use reader.count to check how many records were returned.

To read beyond 10,000 records, enable InstanceTunnel. Choose one of the following options:

Option 1: Enable globally

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

with instance.open_reader() as reader:
    # Reads all data via InstanceTunnel

Option 2: Enable for a single call

with instance.open_reader(tunnel=True, limit=False) as reader:
    # Reads all data for this call only

with o.execute_sql('select * from dual').open_reader() as reader:
    for record in reader:
        # Process each record

with o.execute_sql('desc dual').open_reader() as reader:
    print(reader.raw)

Configure runtime parameters for SQL

Pass runtime parameters to SQL using the hints parameter (type: dict). For supported hint keys, see SET operations.

Per-query hints:

o.execute_sql('select * from PyODPS_iris', hints={'odps.sql.mapper.split.size': 16})

Global hints (applied to all subsequent `execute_sql` calls):

from odps import options
options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('select * from PyODPS_iris')

Use DataFrame to process data

DataFrame is not recommended for new development. Use SQL directly for most data processing tasks.

DataFrame API operations are lazy — they only execute when you call an immediately executed method.

from odps.df import DataFrame
iris = DataFrame(o.get_table('pyodps_iris'))
for record in iris[iris.sepal_width < 3].execute():
    # Process each record

To trigger execution on display (for example, in a notebook-style flow), set options.interactive = True at the top of your script:

from odps import options
from odps.df import DataFrame
options.interactive = True
iris = DataFrame(o.get_table('pyodps_iris'))
print(iris.sepal_width.sum())  # Executes immediately when printed

By default, options.verbose is True in DataWorks, so the Logview URL and other progress details are printed automatically during execution.

End-to-end example

This example loads the pyodps_iris table and prints the first five values in the sepallength column.

1. Prepare the dataset. See DataFrame data processing for how to create the pyodps_iris table.

2. Create a DataFrame object. See Create a DataFrame object from a MaxCompute table.

3. Paste the following code into the node editor and run it:

   from odps.df import DataFrame
   
   # Create a DataFrame object from a MaxCompute table
   iris = DataFrame(o.get_table('pyodps_iris'))
   print(iris.sepallength.head(5))

   Expected output:

      sepallength
   0          4.5
   1          5.5
   2          4.9
   3          5.0
   4          6.0

Install third-party packages

To use a third-party package in a PyODPS node:

1. Use a serverless resource group to run the node.

2. Create a custom image that includes the package.

This method does not apply to third-party packages referenced inside user-defined functions (UDFs). For UDF-specific instructions, see Example: Reference third-party packages in Python UDFs.

Upgrade the PyODPS version

• Serverless resource group: Use the image management feature to run the following command on a PyODPS 3 node: /home/tops/bin/pip3 install pyodps==0.12.1 Replace 0.12.1 with the target version. See Custom images.

• Exclusive resource group for scheduling: Use the O&M Assistant feature to run the same command on a PyODPS 3 node. See O&M Assistant.

Access resources in a VPC or private network

To connect a PyODPS node to a data source in a virtual private cloud (VPC) or private data center, use a serverless resource group and establish a network connection between the resource group and the target. See Network connectivity solutions.

Fix missing data lineages

If SQL statements run in a PyODPS node do not generate data lineages in Data Map, pass DataWorks scheduler runtime parameters as hints when executing SQL:

import os

# Collect DataWorks scheduler runtime parameters
skynet_hints = {}
for k, v in os.environ.items():
    if k.startswith('SKYNET_'):
        skynet_hints[k] = v

# Pass hints when submitting a task
o.execute_sql('INSERT OVERWRITE TABLE XXXX SELECT * FROM YYYY WHERE ***', hints=skynet_hints)

For more on viewing data lineages, see View data lineages. For the hints parameter, see Configure the hints parameter.