: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>]
				

• テーブル列とオプションのパーティションを介して初期化します (以下のコードを参照)。

  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>]
  						

• 初期化は Schema.from_lists を呼び出すほうが簡単ですが、列とパーティションのアノテーションを直接設定することができません。

  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>')
				

• 以下の方法で head を呼び出して、テーブルデータ (取得できるデータレコードは各テーブルの最初の 10,000 個以下) を取得します。

  >>> 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.

• テーブル上で open_reader を実行してリーダーを開き、データを読み取ります。WITH 式を使用できます。

  # 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.

• Tunnel API を呼び出してテーブルデータを読み取ります。 open_reader 操作は、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.

• 基本操作
  テーブルのすべてのパーティションを走査するコードを以下に示します。

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

  パーティションが存在するかどうかチェックするコードを以下に示します。

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

  パーティションを取得するコードを以下に示します。

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

• パーティションの作成

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

• パーティションの削除

   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.

• SQL 文の実行

   odps.execute_sql('select * from dual') # Run SQL in synchronous mode. ブロック化は SQL の実行が完了するまで続きます。
   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.

• SQL 文の実行結果の読み取り
  SQL 文を実行するインスタンスは、直接 open_reader 操作を実行できます。 1 番目のシナリオでは、以下のように SQL 文が構造化データを返します。

  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 タイプ、および py、jar、および archive が含まれます。

  説明 DataWorks では、py 形式のファイルリソースはファイルとしてアップロードする必要があります。 詳細は、「Python UDF」をご参照ください。
  ファイルリソースの作成

  リソース名、ファイルタイプ、およびファイルライクオブジェクト (または文字列オブジェクト) を指定して、ファイルリソースを作成します (以下のコードを参照)。

  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.

  ファイルリソースの読み取りと変更

  ファイルリソースは、open メソッドまたは MaxCompute エントリで open_resource を呼び出してファイルリソースを開くことができます。 開かれたオブジェクトはファイルライクオブジェクトです。 Pythonで構築された open メソッドと同様、ファイルリソースもオープンモードをサポートしています。

  例 :

   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.

  以下のオープンモードをサポートしています。

  • r: 読み取りモードです。 ファイルを開くことはできますが書き込みはできません。
  • w: 書き込みモードです。 ファイルの書き込めはできますが、読み取りはできません。 ファイルを書き込みモードで開くと、最初にファイルの内容が消去されることにご注意ください。
  • a: 追加モードです。 内容をファイルの末尾に追加できます。
  • r+: 読み取り/書き込みモードです。 任意のコンテンツを読み書きできます。
  • w+: r+ とほぼ同じですが、最初にファイルの内容が消去されます。
  • a+: r+ とほぼ同じですが、書き込み中に限りファイルの末尾に内容を追加できます。

  PyODPS では、ファイルリソースはバイナリモードで開くことができます。 たとえば、一部の圧縮ファイルはバイナリモードで開く必要があります。rb はバイナリ読み取りモードでファイルを開くこと、r+b はバイナリ読み取り/書き込みモードでファイルを開くことを示します。

• テーブルリソース
  テーブルリソースの作成

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

  テーブルリソースの更新

  >>> 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'))

• 一般的な設定

  オプション	説明	デフォルト値
  end_point	MaxCompute エンドポイント	なし
  default_project	デフォルトプロジェクト	なし
  log_view_host	LogView ホスト名	なし
  log_view_hours	LogView 保持時間 (1 時間単位)	24
  local_timezone	使用されているタイムゾーン。 True は現地時間、False は UTC を示します。 pytz のタイムゾーンも使用できます。	1
  lifecycle	すべてのテーブルのライフサイクル	なし
  temp_lifecycle	一時テーブルのライフサイクル	1
  biz_id	ユーザー ID	なし
  verbose	ログを印刷するかどうか	False
  verbose_log	ログの受信者	なし
  chunk_size	書き込みバッファのサイズ	1496
  retry_times	再試行のリクエスト回数	4
  pool_connections	接続プール内のキャッシュ接続数	10
  pool_maxsize	接続プールの最大容量	10
  connect_timeout	接続タイムアウト	5
  read_timeout	読み取りタイムアウト	120
  completion_size	オブジェクトの完全なリスト項目の数の上限	10
  notebook_repr_widget	対話型グラフを使用します。	True
  sql.settings	MaxCompute SQL でグローバルヒントを実行します。	なし
  sql.use_odps2_extension	MaxCompute 2.0 の言語拡張を有効にします。	False

• データのアップロード/ダウンロード設定

  オプション	説明	デフォルト値
  tunnel.endpoint	Tunnel エンドポイント	なし
  tunnel.use_instance_tunnel	実行結果を取得するには、Tunnel インスタンスを使用します。	True
  tunnel.limited_instance_tunnel	Tunnel インスタンスによって取得される結果の数を制限します。	True
  tunnel.string_as_binary	文字列型では Unicode ではなくバイトを使用します。	False

• DataFrame の設定

  オプション	説明	デフォルト値
  interactive	インタラクティブ環境にあるかどうか。	検出値に応じます。
  df.analyze	MaxCompute 以外の組み込み関数を有効にするかどうか	True
  df.optimize	DataFrame 全体の最適化を有効にするかどうか	True
  df.optimizes.pp	DataFrame 述語のプッシュ最適化を有効にするかどうか	True
  df.optimizes.cp	DataFrame 列の調整最適化を有効にするかどうか	True
  df.optimizes.tunnel	DataFrameトンネル最適化を有効にするかどうか	True
  df.quote	MaxCompute SQL 末尾のフィールドとテーブル名のマーキングに `` を使うかどうか	True
  df.libraries	実行中の DataFrame に使用されるサードパーティのライブラリ (リソース名)	なし

• PyODPS ML の設定

  オプション	説明	デフォルト値
  ml.xflow_project	デフォルトの Xflow プロジェクト名	algo_public
  ml.use_model_transfer	モデル PMML の取得に ModelTransfer を使用するかどうか。	True
  ml.model_volume	ModelTransfer 使用時に使用されるボリューム名	pyodps_volume