Storage API examples: Python SDK - MaxCompute - Alibaba Cloud Documentation Center

This topic provides code examples for accessing MaxCompute data by using the Python SDK.

MaxCompute provides interfaces for the Storage API. For more information, see aliyun-odps-python-sdk.

Prerequisites

If you run the code in a local environment, ensure PyODPS is installed. For more information, see Install PyODPS.

PyODPS is also available in the following environments:

DataWorks: PyODPS is pre-installed on PyODPS nodes. You can develop and periodically run PyODPS tasks directly on these nodes. For more information, see Use PyODPS in DataWorks.
PAI: You can install and run PyODPS in the PAI Python environment. PyODPS is pre-installed in all built-in PAI images and is ready to use out of the box in components such as the custom Python component in PAI-Designer. Using PyODPS in PAI Notebooks is similar to its standard usage. For more information, see Overview of basic operations and DataFrame (Not recommended).

Note

PyODPS is the Python SDK for MaxCompute. For more information about PyODPS, see PyODPS.

Examples

For complete code examples on how to access MaxCompute by using the Python SDK, see Python SDK Examples.

Configure the environment to connect to the MaxCompute service

import os
from odps import ODPS
from odps.apis.storage_api import *
# Ensure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set to your Access Key ID,
# and the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set to your Access Key Secret.
# For security, avoid hardcoding the Access Key ID and Access Key Secret.
# The endpoint of the MaxCompute service. Only connections from VPC networks are supported.
o = ODPS(
		os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
		os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
		project='your-default-project',
		endpoint='your-end-point'
)
# The name of the MaxCompute table to access.
table = "<table to access>"
# The name of the quota to use for accessing MaxCompute.
quota_name = "<quota name>"
# Connects to the MaxCompute service and creates an Arrow-format Storage API client.
def get_arrow_client():
    odps_table = o.get_table(table)
    client = StorageApiArrowClient(odps=o, table=odps_table, quota_name=quota_name)

    return client

Note

To obtain the quota name for a Storage API exclusive resource group (subscription):

Storage API exclusive resource group: Log on to the MaxCompute console. In the upper-left corner, switch to your region. In the left-side navigation pane, choose Workspace > Quotas to view available quotas. For more information, see Manage quotas for computing resources.
Storage API: Log on to the MaxCompute console. In the left-side navigation pane, choose Tenants > Tenant Property to enable the Storage API.

Read table data

Create a read session to read data from MaxCompute

import logging
import sys
from odps.apis.storage_api import *
from util import *

logger = logging.getLogger(__name__)
# Creates a read session. The mode parameter specifies the split strategy: 'size' to split by data size, or 'row' to split by row offset.
def create_read_session(mode):
    client = get_arrow_client()
    req = TableBatchScanRequest(required_partitions=['pt=test_write_1'])

    if mode == "size":
        req.split_options = SplitOptions.get_default_options(SplitOptions.SplitMode.SIZE)
    elif mode == "row":
        req.split_options = SplitOptions.get_default_options(SplitOptions.SplitMode.ROW_OFFSET)

    resp = client.create_read_session(req)

    if resp.status != Status.OK:
        logger.info("Create read session failed")
        return

    logger.info("Read session id: " + resp.session_id)

if __name__ == '__main__':
    logging.basicConfig(format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', level=logging.INFO)
    if len(sys.argv) != 2:
        raise ValueError("Please provide split mode: size|row")

    mode = sys.argv[1]
    if mode != "row" and mode != "size":
        raise ValueError("Please provide split mode: size|row")

    create_read_session(mode)

Check the status of a read session

import logging
import sys
import time
from odps.apis.storage_api import *
from util import *

logger = logging.getLogger(__name__)
# Before reading data, ensure the read session is created and ready.
def check_session_status(session_id):
    client = get_arrow_client()
    req = SessionRequest(session_id=session_id)
    resp = client.get_read_session(req)

    if resp.status != Status.OK:
        logger.info("Get read session failed")
        return

    # Session creation can be time-consuming. You must wait for the session status to become NORMAL before reading data.
    if resp.session_status == SessionStatus.NORMAL:
        logger.info("Read session id: " + resp.session_id)
    else:
        logger.info("Session status is not expected")

if __name__ == '__main__':
    logging.basicConfig(format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', level=logging.INFO)
    if len(sys.argv) != 2:
        raise ValueError("Please provide session id")

    session_id = sys.argv[1]
    check_session_status(session_id)

Read MaxCompute data

# Reads data rows from MaxCompute for a specified session_id and counts the total number of rows.
import logging
import sys
from odps.apis.storage_api import *
from util import *

logger = logging.getLogger(__name__)

def read_rows(session_id):
    client = get_arrow_client()
    req = SessionRequest(session_id=session_id)
    resp = client.get_read_session(req)

    if resp.status != Status.OK and resp.status != Status.WAIT:
        logger.info("Get read session failed")
        return

    req = ReadRowsRequest(session_id=session_id)
    if resp.split_count == -1:
        req.row_index = 0
        req.row_count = resp.record_count
    else:
        req.split_index = 0

    reader = client.read_rows_arrow(req)
    total_line = 0
    while True:
        record_batch = reader.read()
        if record_batch is None:
            break
        total_line += record_batch.num_rows

    if reader.get_status() != Status.OK:
        logger.info("Read rows failed")
        return

    logger.info("Total line is:" + str(total_line))

if __name__ == '__main__':
    logging.basicConfig(format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', level=logging.INFO)
    if len(sys.argv) != 2:
        raise ValueError("Please provide session id")

    session_id = sys.argv[1]
    read_rows(session_id)

MaxCompute:Open Storage Python SDK

Prerequisites

Examples

Related documents