MaxCompute:Develop a PyODPS 3 task

Prerequisites

Before you begin, ensure that you have:

• Created a PyODPS 3 node. For more information, see Create and manage MaxCompute nodes

Write and run node code

PyODPS is MaxCompute SDK for Python. In DataWorks, you can use PyODPS nodes to schedule Python tasks and integrate Python tasks with other types of tasks.

After creating a PyODPS 3 node, write and run Python code directly in the node editor. For the full PyODPS API reference, see Overview.

print(odps.exist_table('PyODPS_iris'))

options.tunnel.use_instance_tunnel = True
options.tunnel.limit_instance_tunnel = False  # Remove the record limit

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

# Enable InstanceTunnel and remove the record limit for this call only
with instance.open_reader(tunnel=True, limit=False) as reader:
    ...

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

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

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)

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

from odps import options
from odps.df import DataFrame
options.interactive = True  # Set at the top of the script
iris = DataFrame(o.get_table('pyodps_iris'))
print(iris.sepal_width.sum())  # Executes immediately

Schedule a node

To run a PyODPS 3 node on a recurring schedule, configure scheduling properties in the node's Properties tab. For an overview of scheduling options, see Overview.

print('ds=' + args['ds'])
# Output: ds=20161116

o.get_table('table_name').get_partition('ds=' + args['ds'])

Limitations

Additional constraints:

• Memory limits do not apply to SQL or DataFrame tasks (excluding to_pandas tasks) initiated by PyODPS.

• NumPy and pandas are pre-installed in DataWorks for use in non-UDF code. Third-party packages that contain binary code are not supported.

• options.tunnel.use_instance_tunnel is set to False by default in DataWorks. To enable InstanceTunnel globally, set it to True.

• MaxCompute is compatible with Python 3.7. Other Python 3 subversions may fail on certain syntax — for example, Python 3.8 fails on finally block syntax. Use Python 3.7 to avoid compatibility errors.

Usage notes

• Third-party packages: To use a third-party package in a PyODPS node running on a DataWorks resource group, use a serverless resource group and create a custom image to install the package.

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

• Upgrading PyODPS: Run /home/tops/bin/pip3 install pyodps==0.12.1 on the resource group, replacing 0.12.1 with the target version.

  • Serverless resource group: use the image management feature. For more information, see Custom images.

  • Exclusive resource group for scheduling: use the O&M Assistant feature. For more information, see O&M Assistant.

• VPC and data center access: To access a data source or service in a virtual private cloud (VPC) or data center, run the node on a serverless resource group and establish a network connection between the resource group and the target. For more information, see Network connectivity solutions.

• Data lineage: If SQL statements in a PyODPS node cannot generate data lineages and Data Map does not display them correctly, manually configure the scheduling parameters in node code using the hints mechanism. For more information, see View data lineages and Configure the hints parameter. The following example shows how to get the parameters required for running tasks:

  import os
  ...
  # Get 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)
  ...

• PyODPS version: PyODPS 2 nodes use Python 2; PyODPS 3 nodes use Python 3. Choose the version that matches the Python version in your code. For the full API reference, see PyODPS.

What's next

• Determine whether a custom Shell script is successfully run: The logic applies equally to Python scripts.

• Deploy the PyODPS 3 node: In a workspace running in standard mode, deploy the node to the production environment before it can be scheduled.

• Perform O&M on the PyODPS 3 node: After deploying to Operation Center, manage the node's operational health from Operation Center.

• Learn FAQ about PyODPS: Find answers to common issues to speed up troubleshooting.

FAQ

When I run a PyODPS 3 node in Operation Center to collect data from a third-party app (such as Lark), the node times out — even though it runs successfully on my local machine. What's wrong?

The node is running in a sandbox that blocks outbound connections to external addresses by default. In the DataWorks console, go to Management Center > Workspace > Security Settings, and add the third-party app's address to the sandbox whitelist.