MaxCompute:テーブルの管理

カラムおよびパーティションを明示的に定義

Schema、Column、Partition を odps.models からインポートし、カラムおよびパーティションのリストを Schema コンストラクターに渡します。このアプローチでは、各カラムおよびパーティションに対して comment パラメーターを指定できます。

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

スキーマを作成した後、以下のプロパティでその内容を確認できます。

• すべてのカラム（パーティションキー列を含む）：出力結果：

    print(schema.columns)

    [<column num, type bigint>,
     <column num2, type double>,
     <partition pt, type string>]

• パーティションキー列のみ：出力結果：

    print(schema.partitions)

    [<partition pt, type string>]

• パーティション以外のカラム名：出力結果：

    print(schema.names)

    ['num', 'num2']

• パーティション以外のカラムのデータの型：出力結果：

    print(schema.types)

    [bigint, double]

Schema.from_lists() を使用

Schema.from_lists() は、名前のリストとデータの型のリストを受け取る簡略化されたメソッドです。この方法はシンプルですが、カラムやパーティションに対するコメントを直接指定することはできません。

from odps.models import Schema
schema = Schema.from_lists(['num', 'num2'], ['bigint', 'double'], ['pt'], ['string'])
print(schema.columns)

出力結果：

[<column num, type bigint>,
 <column num2, type double>,
 <partition pt, type string>]

スキーマからテーブルを作成

まず Schema を構築し、それを create_table() に渡します。

# テーブルスキーマを作成します。
from odps.models import Schema
schema = Schema.from_lists(['num', 'num2'], ['bigint', 'double'], ['pt'], ['string'])

# 作成したスキーマを使用してテーブルを作成します。
table = o.create_table('my_new_table', schema)

# 同じ名前のテーブルが存在しない場合にのみテーブルを作成します。
table = o.create_table('my_new_table', schema, if_not_exists=True)

# テーブルのライフサイクルを設定します。
table = o.create_table('my_new_table', schema, lifecycle=7)

テーブルが正しく作成されたかを確認します。

print(o.exist_table('my_new_table'))

True が返された場合、テーブルは正常に作成されています。

カラム定義からテーブルを作成

カラム名およびデータの型を文字列またはタプルとして create_table() に渡します。

# 指定された共通カラムおよびパーティションキー列を持つパーティションテーブル my_new_table を作成します。
table = o.create_table('my_new_table', ('num bigint, num2 double', 'pt string'), if_not_exists=True)

# パーティションを持たないテーブル my_new_table02 を作成します。
table = o.create_table('my_new_table02', 'num bigint, num2 double', if_not_exists=True)

テーブルが正しく作成されたかを確認します。

print(o.exist_table('my_new_table'))

True が返された場合、テーブルは正常に作成されています。

MaxCompute V2.0 拡張データの型を有効化

デフォルトでは、create_table() は BIGINT、DOUBLE、DECIMAL、STRING、DATETIME、BOOLEAN、MAP、ARRAY のみをサポートします。TINYINT や STRUCT などの追加のデータの型を使用するには、options.sql.use_odps2_extension を 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>')

テーブルがパーティション化されているかを確認

table = o.get_table('my_new_table')
if table.schema.partitions:
    print('Table %s is partitioned.' % table.name)

パーティションを反復処理

table = o.get_table('my_new_table')
for partition in table.partitions:  # すべてのパーティションを反復処理します。
    print(partition.name)  # 反復処理のステップです。このステップではパーティション名が表示されます。
for partition in table.iterate_partitions(spec='pt=test'):  # test という名前のパーティション内のレベル 2 のパーティションを反復処理します。
    print(partition.name)  # 反復処理のステップです。このステップではパーティション名が表示されます。
for partition in table.iterate_partitions(spec='dt>20230119'):  # dt>20230119 条件を満たすパーティション内のレベル 2 のパーティションを反復処理します。
    print(partition.name)  # 反復処理のステップです。このステップではパーティション名が表示されます。

パーティションの存在確認

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

パーティション情報の取得

table = o.get_table('my_new_table')
partition = table.get_partition('pt=test')
print(partition.creation_time)
partition.size

パーティションの作成

t = o.get_table('my_new_table')
t.create_partition('pt=test', if_not_exists=True)  # 同じ名前のパーティションが存在しない場合にのみパーティションを作成します。

パーティションの削除

t = o.get_table('my_new_table')
t.delete_partition('pt=test', if_exists=True)  # if_exists パラメーターを True に設定します。これにより、パーティションが存在する場合にのみ削除されます。
partition.drop()  # drop() メソッドを呼び出して、パーティションが存在する場合に削除します。

head() を使用して先頭 N 件のレコードを読み取り

head() メソッドは、テーブルから最大 10,000 件までの先頭レコードを取得します。

from odps import ODPS
t = o.get_table('dual')
for record in t.head(3):
    # 各レコードを処理します。

open_reader() を with 文とともに使用して読み取り

with t.open_reader(partition='pt=test') as reader:
count = reader.count
for record in reader[5:10]  # 全レコードを読み取るまでこの文を複数回実行できます。読み取るレコード数は count で指定されます。並列処理用のコードに変更することもできます。
    # 1 件のレコードを処理します。

open_reader() を with 文なしで使用して読み取り

reader = t.open_reader(partition='pt=test')
count = reader.count
for record in reader[5:10]  # 全レコードを読み取るまでこの文を複数回実行できます。読み取るレコード数は count で指定されます。並列処理用のコードに変更することもできます。
    # 1 件のレコードを処理します。

pandas DataFrame への直接読み取り

with t.open_reader(partition='pt=test') as reader:
pd_df = reader.to_pandas()

open_writer() を with 文とともに使用して書き込み

with t.open_writer(partition='pt=test') as writer:
	  records = [[111, 'aaa', True],                 # リストを使用できます。
	             [222, 'bbb', False],
	             [333, 'ccc', True],
	             [444, 'Chinese', False]]
    writer.write(records)  # レコードはイテラブルオブジェクトにすることができます。

records = [t.new_record([111, 'aaa', True]),   # レコードオブジェクトを使用できます。
           t.new_record([222, 'bbb', False]),
           t.new_record([333, 'ccc', True]),
           t.new_record([444, 'Chinese', False])]
writer.write(records)

Record オブジェクト（t.new_record() で作成）または単純なリストのいずれかを writer.write() に渡すことができます。

書き込み時にパーティションを自動作成

パーティションが存在しない場合に自動的に作成するには、create_partition=True を設定します。

with t.open_writer(partition='pt=test', create_partition=True) as writer:
    records = [[111, 'aaa', True],                 # リストを使用できます。
               [222, 'bbb', False],
               [333, 'ccc', True],
               [444, 'Chinese', False]]
    writer.write(records)  # レコードはイテラブルなオブジェクトでなければなりません。

write_table() を使用して書き込み

MaxCompute エントリーオブジェクトの write_table() メソッドは、データ書き込みのためのよりシンプルなインターフェイスを提供します。

records = [[111, 'aaa', True],                 # リストを使用できます。
           [222, 'bbb', False],
           [333, 'ccc', True],
           [444, 'Chinese', False]]
o.write_table('test_table', records, partition='pt=test', create_partition=True)