Python SDK

from odps import ODPS
odps = ODPS('**your-access-id**', '**your-secret-access-key**', '**your-default-project**',
            endpoint='**your-end-point**')

project = odps.get_project('my_project') # Obtain a project.
project = odps.get_project() # Obtain the default project.

for table in odps.list_tables():
    # Process each table

t = odps.get_table('dual')
t.schema
odps.Schema {
  c_int_a                 bigint
  c_int_b bigint
  c_double_a double
  c_double_b double
  c_string_a string
  c_string_b string
  c_bool_a boolean
  c_bool_b boolean
  c_datetime_a datetime
  c_datetime_b datetime
}
t.lifecycle
-1
print(t.creation_time)
2014-05-15 14:58:43
t.is_virtual_view
False
t.size
1408
t.schema.columns
[<column c_int_a, type bigint>,
 <column c_int_b, type bigint>,
 <column c_double_a, type double>,
 <column c_double_b, type double>,
 <column c_string_a, type string>,
 <column c_string_b, type string>,
 <column c_bool_a, type boolean>,
 <column c_bool_b, type boolean>,
 <column c_datetime_a, type datetime>,
 <column c_datetime_b, type datetime>]

• Initialize through table columns and optional partitions, as shown in the following code:

  from odps.models import Schema, Column, Partition
  columns = [Column(name='num', type='bigint', comment='the column')]
  partitions = [Partition(name='pt', type='string', comment='the partition')]
  schema = Schema(columns=columns, partitions=partitions)
  schema.columns
  [<column num, type bigint>, <partition pt, type string>]
  

• Although it is easier to call Schema.from_lists for initialization, annotations of columns and partitions cannot be set directly.

  schema = Schema.from_lists(['num'], ['bigint'], ['pt'], ['string'])
  schema.columns
  [<column num, type bigint>, <partition pt, type string>]
  

table = odps.create_table('my_new_table', schema)
table = odps.create_table('my_new_table', schema, if_not_exists=True) # Create a table only when no table exists.
table = o.create_table('my_new_table', schema, lifecycle=7) # Set the life cycle.

>>> # Create a non-partition table.
>>> table = o.create_table('my_new_table', 'num bigint, num2 double', if_not_exists=True)
>>> # To create a partition table, you can input (list of table fields, list of partition fields).
>>> table = o.create_table('my_new_table', ('num bigint, num2 double', 'pt string'), if_not_exists=True)

from odps import options
options.sql.use_odps2_extension = True
table = o.create_table('my_new_table', 'cat smallint, content struct<title:varchar(100), body string>')

• Call head to obtain table data as follows (only the first 10,000 data records or fewer of each table can be obtained):

  >>> t = odps.get_table('dual')
  >>> for record in t.head(3):
  >>>     print(record[0]) # Obtain the value at the zero position.
  >>>     print(record['c_double_a']) # Obtain a value through a field.
  >>>     print(record[0: 3]) # Slice action
  >>>     print(record[0]) # Obtain values at multiple positions.
  >>>     print(record['c_int_a', 'c_double_a']) # Obtain values through multiple fields.

• Run open_reader on a table to open a reader to read data. You can use the WITH expression:

  # Use the with expression.
  with t.open_reader(partition='pt=test') as reader:
      count = reader.count
      for record in reader[5:10] # This action can be performed multiple times until a certain number (indicated by count) of records are read. This statement can be transformed to parallel action.
          # Process a record.
  
   # Do not use the with expression.
   reader = t.open_reader(partition='pt=test')
   count = reader.count
   for record in reader[5:10]
       # Process a record.

• Call the Tunnel API to read table data. The open_reader action is encapsulated in the Tunnel API.

 # Use the with expression.
 with t.open_writer(partition='pt=test') as writer:
     writer.write(records) # Here, records can be any iteratable records and are written to block 0 by default.

 with t.open_writer(partition='pt=test', blocks=[0, 1]) as writer: # Open two blocks at the same time
     writer.write(0, gen_records(block=0))
     writer.write(1, gen_records(block=1)) # The two write operations can be parallel in multiple threads. Each block is independent.

 # Do not use the WITH expression.
 writer = t.open_writer(partition='pt=test', blocks=[0, 1])
 writer.write(0, gen_records(block=0))
 writer.write(1, gen_records(block=1))
 writer.close() # You must close the writer. Otherwise, the written data may be incomplete.

odps.delete_table('my_table_name', if_exists=True) # Delete a table only when the table exists
 t.drop() # The drop function can be directly executed if a table object exists.

• Basic operations
  Traverse all partitions of a table as shown in the following code:

  for partition in table.partitions:
      print(partition.name)
   for partition in table.iterate_partitions(spec='pt=test'):
      Traverse list partitions.

  Check whether a partition exists as shown in the following code:

   table.exist_partition('pt=test,sub=2015')

  Obtain the partition as shown in the following code:

   partition = table.get_partition('pt=test')
   print(partition.creation_time)
  2015-11-18 22:22:27
   partition.size
  0

• Create a partition

   t.create_partition('pt=test', if_not_exists=True) # Create a partition only when no partition exists.

• Delete a partition

   t.delete_partition('pt=test', if_exists=True) # Delete a partition only when the partition exists.
   partition.drop() # Directly drop a partition if a partition object exists.

• Run the SQL statements

   odps.execute_sql('select * from dual') # Run SQL in synchronous mode. Blocking continues until SQL execution is completed.
   instance = odps.run_sql('select * from dual') # Run the SQL statements in asynchronous mode.
  instance.wait_for_success() # Blocking continues until SQL execution is completed.

• Read the SQL statement execution results
  The instance that runs the SQL statements can directly perform the open_reader action. In one scenario, the SQL statements return structured data, as follows:

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

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

• File resources
  File resources include the basic file type, and py, jar, and archive.

  Note In DataWorks, file resources in the py format must be uploaded as files. For more information, see Python UDF.
  Create a file resource

  Create a file resource by specifying the resource name, file type, and a file-like object (or a string object), as shown in the following example:

  resource = odps.create_resource('test_file_resource', 'file', file_obj=open('/to/path/file')) # Use a file-like object.
  resource = odps.create_resource('test_py_resource', 'py', file_obj='import this') # Use a string.

  Read and modify a file resource

  You can call the open method for a file resource or call open_resource at the MaxCompute entry to open a file resource. The opened object is a file-like object. Similar to the open method built in Python, file resources also support the open mode. 

  Example:

   with resource.open('r') as fp: # Open a resource in read mode.
      content = fp.read() # Read all content.
      fp.seek(0) # Return to the start of the resource.
      lines = fp.readlines() # Read multiple lines.
      fp.write('Hello World') # Error. Resources cannot be written in read mode.
  with odps.open_resource('test_file_resource', mode='r+') as fp: # Enable read/write mode.
      fp.read()
      fp.tell() # Current position
      fp.seek(10)
      fp.truncate() # Truncate the following content.
      fp.writelines(['Hello\n', 'World\n']) # Write multiple lines.
      fp.write('Hello World')
      fp.flush() # Manual call submits the update to MaxCompute.

  The following open modes are supported:

  • r: Read mode. The file can be opened but cannot be written.
  • w: Write mode. The file can be written but cannot be read. Note that file content is cleared first if the file is opened in write mode.
  • a: Append mode. Content can be added to the end of the file.
  • r+: Read/write mode. You can read and write any content.
  • w+: Similar to r+, but file content is cleared first.
  • a+: Similar to r+, but content can be added at the end of the file only during writing.

  In PyODPS, file resources can be opened in a binary mode. For example, some compressed files must be opened in binary mode.  rb indicates opening a file in binary read mode, and r+b indicates opening a file in binary read/write mode.

• Table resources
  Create a table resource

  >>> odps.create_resource('test_table_resource', 'table', table_name='my_table', partition='pt=test')

  Update a table resource

  >>> table_resource = odps.get_resource('test_table_resource')
  >>> table_resource.update(partition='pt=test2', project_name='my_project2')

o = ODPS('**your-access-id**', '**your-secret-access-key**',
             project='**your-project**', endpoint='**your-end-point**'))

from odps.df import DataFrame

users = DataFrame(o.get_table('pyodps_ml_100k_users'))

users.dtypes

users.head(10)

users[['user_id', 'age']].head(5)

users.exclude('zip_code', 'age').head(5)

users.select(users.exclude('zip_code', 'sex'), sex_bool=users.sex == 'M').head(5)

users.age.between(20, 25).count().rename('count')
943

users.groupby(users.sex).count()

>>> df = users.groupby('occupation').agg(count=users['occupation'].count())
>>> df.sort(df['count'], ascending=False)[:10]

users.occupation.value_counts()[:10]

%matplotlib inline

users['occupation'].value_counts().plot(kind='barh', x='occupation', 
ylabel='prefession')

users.age.hist(bins=30, title="Distribution of users' ages", xlabel='age', ylabel='count of users')

movies = DataFrame(o.get_table('pyodps_ml_100k_movies'))
ratings = DataFrame(o.get_table('pyodps_ml_100k_ratings'))
o.delete_table('pyodps_ml_100k_lens', if_exists=True)
 lens = movies.join(ratings).join(users).persist('pyodps_ml_100k_lens')
 lens.dtypes

odps.Schema {
  movie_id int64
  title string
  release_date string
  video_release_date string
  imdb_url string
  user_id int64
  rating int64
  unix_timestamp int64
  age int64
  sex string
  occupation string
  zip_code string
}

 labels = ['0-9', '10-19', '20-29', '30-39', '40-49', '50-59', '60-69', '70-79']
 cut_lens = lens[lens, lens.age.cut(range(0, 81, 10), right=False, labels=labels).rename('age group')]

>>> cut_lens['age group', 'age'].distinct()[:10]

cut_lens.groupby('age group').agg(cut_lens.rating.count().rename('total rating'), cut_lens.rating.mean().rename('average rating'))

• General configuration

  Option	Description	Default value
  end_point	MaxCompute Endpoint.	None
  default_project	Default Project.	None
  log_view_host	LogView host name.	None
  log_view_hours	LogView holding time (in hours).	24
  local_timezone	Used time zone. True indicates local time, and False indicates UTC. The time zone of pytz can also be used.	1
  lifecycle	Life cycles of all tables.	None
  temp_lifecycle	Life cycles of the temporary tables.	1
  biz_id	User ID.	None
  verbose	Whether to print logs.	False
  verbose_log	Log receiver.	None
  chunk_size	Size of write buffer.	1496
  retry_times	Request retry times.	4
  pool_connections	Number of cached connections in the connection pool.	10
  pool_maxsize	Maximum capacity of the connection pool.	10
  connect_timeout	Connection time-out.	5
  read_timeout	Read time-out.	120
  completion_size	Limit on the number of object complete listing items.	10
  notebook_repr_widget	Use interactive graphs.	True
  sql.settings	MaxCompute SQL runs global hints.	None
  sql.use_odps2_extension	Enable MaxCompute 2.0 language extension.	False

• Data Upload/Download configuration

  Option	Description	Default value
  tunnel.endpoint	Tunnel Endpoint.	None
  tunnel.use_instance_tunnel	Use Instance Tunnel to obtain the execution result.	True
  tunnel.limited_instance_tunnel	Limit the number of results obtained by Instance Tunnel.	True
  tunnel.string_as_binary	Use bytes instead of unicode in the string type.	False

• DataFrame Configurations

  Option	Description	Default value
  interactive	Whether in an interactive environment.	Depend on the detection value
  df.analyze	Whether to enable non-MaxCompute built-in functions.	True
  df.optimize	Whether to enable DataFrame overall optimization.	True
  df.optimizes.pp	Whether to enable DataFrame predicate push optimization.	True
  df.optimizes.cp	Whether to enable DataFrame column tailoring optimization.	True
  df.optimizes.tunnel	Whether to enable DataFrame tunnel optimization.	True
  df.quote	Whether to use `` to mark fields and table names at the end of MaxCompute SQL.	True
  df.libraries	Third-party library (resource name) that is used for DataFrame running.	None

• PyODPS ML Configurations

  Option	Description	Default value
  ml.xflow_project	Default Xflow project name.	algo_public
  ml.use_model_transfer	Whether to use ModelTransfer to obtain the model PMML.	True
  ml.model_volume	Volume name used when ModelTransfer is used.	pyodps_volume