repository·master·Indexed 19 days ago
https://github.com/aliyun/aliyun-odps-python-sdkA Python SDK and data analysis framework for Alibaba Cloud MaxCompute (ODPS). It enables developers to manage tables, execute SQL queries, and develop/debug User Defined Functions (UDFs). The SDK includes a DataFrame API (odps.df) for distributed data processing, high-performance data access via Storage API V2 and TableTunnel, and an IPython/Jupyter extension for running SQL via %sql magic.
A Schema is a MaxCompute concept that sits between a Project and objects like Tables, Resources, or Functions. It is used to further categorize these objects.
Note: Schema is a MaxCompute public beta feature. You must apply for access via the official documentation. Using Schema requires PyODPS version 0.11.3 or higher.
When using the PyODPS IPython extension, you can use the Tab key to trigger automatic completion for ODPS objects. This works for:
o.get_table(<tab>)o.get_table('tabl<tab>')o.get_table(project='project_name', name='tabl<tab>') or o.get_table('tabl<tab>', project='project_name')).If multiple objects match, IPython will display a list. The maximum number of suggestions is controlled by options.completion_size (default is 10).
DataWorks imposes CPU and memory limits on PyODPS nodes to protect the gateway.
Got killed, it indicates the process exceeded memory limits. Avoid performing large-scale data operations locally in the node..to_pandas()) which run on MaxCompute and are not subject to these local gateway limits.numpy. Binary-based third-party packages are not supported.The odps.df module provides a DataFrame API for performing distributed data processing on MaxCompute (ODPS). It allows you to define data transformations using a syntax similar to Pandas, which are then executed as optimized SQL/MaxCompute jobs.
Key components include:
odps.df.DataFrame: The primary entry point for representing and manipulating distributed datasets.odps.df.expr classes: Used for defining complex expressions and sequences within the DataFrame context.odps.df.GroupBy: Provides grouping and aggregation capabilities for DataFrame operations.CollectionExpr class. A CollectionExpr can be thought of as an ODPS table or an electronic form. A DataFrame object is a special type of CollectionExpr. CollectionExpr provides a wide range of operations for column manipulation, filtering, and transformations on two-dimensional datasets.MaxCompute supports composite types. You can create them using constructors or type strings.
odps_types.Array(value_type)"array<type>".value_type to get the element type.odps_types.Map(key_type, value_type)"map<key_type, value_type>".key_type and .value_type.odps_types.Struct([("name", type), ...]) or odps_types.Struct({"name": type, ...})"struct<field1:type1, field2:type2>".field_types to get an OrderedDict of field names and their types.import odps.types as odps_types
from odps.types import validate_data_type
# Array
array_type = odps_types.Array(odps_types.bigint)
# Map
map_type = odps_types.Map(odps_types.string, odps_types.Array(odps_types.bigint))
# Struct
struct_type = odps_types.Struct([("a", odps_types.bigint), ("b", odps_types.string)])
# Using strings for all
array_str = validate_data_type("array<bigint>")
map_str = validate_data_type("map<string, array<bigint>>")
struct_str = validate_data_type("struct<a:bigint, b:string>")
# Inspecting Struct fields
for field_name, field_type in struct_str.field_types.items():
print(f"field_name: {field_name} field_type: {field_type}")PyODPS provides support for both the DBAPI and SQLAlchemy standards. This allows you to operate on MaxCompute using standard Python database interfaces, enabling seamless interoperability with third-party libraries and environments that expect these common protocols.
To use these interfaces, you can follow the specific guides for either DBAPI or SQLAlchemy provided in the PyODPS documentation.
When working with PyODPS DataFrame, you interact with three primary object types:
DataFrame): Represents a 2D data structure (a table). It serves as a reference to a data source.SequenceExpr): Represents a 1D structure (a column) within a Collection.Important Note on Execution: When these objects are created from ODPS tables or partitions, they do not contain actual data in memory. They only contain the operations to be performed. The actual storage and computation occur on the ODPS side. Actual data is only loaded into local memory after creating a DataFrame from a Pandas object or performing an action that triggers data retrieval.
Storage API V2 is a high-throughput data read/write interface for MaxCompute. It provides fine-grained session management and supports Arrow and Blob formats. Key abstractions include:
UNCOMPRESSED (default), LZ4, and ZSTD algorithms.access_token and row_offset.PyODPS supports both asynchronous and parallel execution for immediate methods (execute, persist, head, tail, to_pandas).
Pass async_=True to an immediate method. It returns a concurrent.futures.Future object. Use .result() to wait for and retrieve the result.
Pass n_parallel=N to execute() to specify the concurrency level. This is effective when a DataFrame's execution depends on multiple cached DataFrames that can be computed in parallel.
To avoid redundant computations when multiple expressions share common dependencies, use the Delay API. You register operations with a Delay object and then call delay.execute(n_parallel=N). This automatically identifies common dependencies and executes them once before running the dependent tasks in parallel.
# Asynchronous execution
future = iris[iris.sepal_width < 10].head(10, async_=True)
print(future.result())
# Parallel execution with multiple dependencies
expr1 = iris.groupby('category').agg(value=iris.sepal_width.sum()).cache()
expr2 = iris.groupby('category').agg(value=iris.sepal_length.mean()).cache()
expr = expr1.union(expr2)
future = expr.execute(n_parallel=2, async_=True, timeout=2)
print(future.result())
# Using the Delay API for optimized dependency management
from odps.df import Delay
delay = Delay()
df = iris[iris.sepal_width < 5].cache() # Common dependency
# These return futures immediately without executing
future1 = df.sepal_width.sum().execute(delay=delay)
future2 = df.sepal_width.mean().execute(delay=delay)
# Trigger execution with specified concurrency
delay.execute(n_parallel=3)
print(future1.result())PyODPS provides a DataFrame API that offers an interface similar to pandas. This allows you to write code that can leverage ODPS's distributed computing power or run locally using pandas for computation.
Important Limitations:
Series support, Index support, row-based reading, or horizontal merging of multiple DataFrames using iloc.Calling .to_pandas() often results in memory usage significantly higher than the table size on disk due to:
Tips:
df.memory_usage(deep=True).sum() in Pandas for a more accurate memory measurement.