MaxCompute:Python SDK examples

Step 1: Set up the connection

Create an ODPS client and an Apache Arrow-format Storage API client. All subsequent read operations use the get_arrow_client() function defined here.

import os
from odps import ODPS
from odps.apis.storage_api import *

# Load AccessKey credentials from environment variables.
# Do not hardcode credentials in your source code.
o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='<your-endpoint>'
)

# Name of the MaxCompute table to read
table = '<table-name>'
# Quota name for the Storage API exclusive resource group (subscription)
quota_name = '<quota-name>'

def get_arrow_client():
    odps_table = o.get_table(table)
    client = StorageApiArrowClient(odps=o, table=odps_table, quota_name=quota_name)
    return client

Replace the following placeholders:

To find your quota name, log on to the MaxCompute console, switch to your region, and go to Workspace > Quotas in the left-side navigation pane. See Manage quotas for computing resources in the MaxCompute console for details.

To enable the Storage API, go to Tenants > Tenant Property in the left-side navigation pane and turn on Storage API Switch.

Step 2: Create a read session

Submit a TableBatchScanRequest to start a read session. The split mode controls how MaxCompute partitions the data for parallel reads: SIZE splits by data volume, ROW_OFFSET splits by number of rows.

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

logger = logging.getLogger(__name__)

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 not in ('row', 'size'):
        raise ValueError('Please provide split mode: size|row')

    create_read_session(mode)

Copy the session_id from the log output — you need it in the next two steps.

Step 3: Verify the session status

Session creation is asynchronous and may take some time. Poll get_read_session until the status is SessionStatus.NORMAL before reading data.

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

logger = logging.getLogger(__name__)

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

    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)

When the log shows the session ID, the session is ready.

Step 4: Read data rows

Use the session ID to fetch data in Apache Arrow record batches. The script counts total rows as a basic verification.

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:
        # No splits: read all rows in a single range
        req.row_index = 0
        req.row_count = resp.record_count
    else:
        # Splits available: read from split 0
        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)

When the read succeeds, the log prints the total number of rows read from the table.

The example above reads from split_index = 0. When split_count is greater than 0, the session contains multiple splits that can be distributed across threads or worker nodes. Iterate over all split indices (0 through split_count - 1) and assign each to a separate reader for parallel ingestion.