Platform For AI:MaxCompute の使用

準備：アカウント情報の設定

PAIIO を使用して MaxCompute テーブルからデータを読み取る、または MaxCompute テーブルにデータを書き込む前に、MaxCompute リソースにアクセスするための AccessKey 情報を設定する必要があります。PAI では、設定ファイルから AccessKey 情報を取得できます。これを実現するには、設定ファイルをファイルシステムに保存し、環境変数を使用してコード内でその情報を参照します。

1. 次の内容で設定ファイルを作成します。

   access_id=xxxx
   access_key=xxxx
   end_point=http://xxxx

   パラメーター	説明
   access_id	Alibaba Cloud アカウントの AccessKey ID です。
   access_key	Alibaba Cloud アカウントの AccessKey Secret です。
   end_point	MaxCompute のエンドポイントです。例えば、China (Shanghai) リージョンのエンドポイントは http://service.cn-shanghai.maxcompute.aliyun.com/api です。詳細については、「エンドポイント」をご参照ください。

2. 次の構文で、コード内に設定ファイルのパスを指定します。

   os.environ['ODPS_CONFIG_FILE_PATH'] = '<your MaxCompute config file path>'

   <your MaxCompute config file path> には、実際のファイルパスを設定します。

TableRecordDataset

概要

オープンソースの TensorFlow では、TensorFlow 1.2 以降において、従来のスレッディングおよびキューイングのインターフェイスに代わり TensorFlow データセット を使用してデータストリームを作成することを推奨します。複数の Dataset インターフェイスを組み合わせて、計算用のデータを生成します。これにより、データ入力のコードが簡素化されます。

• 構文 (Python)

  class TableRecordDataset(Dataset):
    def __init__(self,
                 filenames,
                 record_defaults,
                 selected_cols=None,
                 excluded_cols=None,
                 slice_id=0,
                 slice_count=1,
                 num_threads=0,
                 capacity=0):

• パラメーター

  パラメーター	必須	タイプ	デフォルト値	説明
  filenames	はい	STRING	なし	読み取り対象のテーブル名です。テーブルは同一のスキーマを使用している必要があります。テーブル名の形式：odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/...。
  record_defaults	はい	LIST または TUPLE	なし	各列のデフォルト値を指定するリストまたはタプルです。このパラメーターは、データ型の推論と欠損値の補完に使用されます。列が空の場合、ここで指定された値が使用されます。値のデータ型が列のデータ型と一致しない場合や、自動的に変換できない場合は、システムが例外をスローします。 サポートされるデータ型は FLOAT32、FLOAT64、INT32、INT64、BOOL、STRING です。INT64 データ型のデフォルト値については、np.array(0, np.int64) 構文を使用します。
  selected_cols	いいえ	STRING	なし	選択する列。複数の列はコンマ (,) で区切ります。このパラメーターをデフォルト値の [None] に設定すると、すべての列が読み取られます。[selected_cols] と [excluded_cols] は、いずれか一方のみ指定できます。
  excluded_cols	いいえ	STRING	なし	除外する列です。複数の列はカンマ (,) で区切ります。このパラメーターをデフォルト値の None に設定することは、除外する列がないことを意味します。selected_cols と excluded_cols のいずれか一方のみを指定できます。
  slice_id	いいえ	INT	0	分散読み取りモードにおけるシャードの ID です。シャード ID は 0 から始まります。分散読み取りモードでは、テーブルは slice_count パラメーターの値に基づいて複数のシャードに分割されます。システムは、slice_id パラメーターで指定したシャードからデータを読み取ります。 slice_id をデフォルト値の 0 に設定し、slice_count を 1 に設定した場合は、テーブル全体が読み取られます。slice_id をデフォルト値の 0 に設定し、slice_count を 1 より大きい値に設定した場合は、0 番目のシャードが読み取られます。
  slice_count	いいえ	INT	1	分散読み取りモードにおけるシャード数です。ほとんどの場合、この値はワーカー数と一致します。このパラメーターをデフォルト値の 1 に設定した場合、シャーディングせずにテーブル全体が読み取られます。
  num_threads	いいえ	INT	0	各テーブルの組み込みリーダーがデータをプリフェッチするために使用するスレッド数です。これらのスレッドは計算スレッドとは独立しています。有効な値：1～64。num_threads を 0 に設定した場合、システムは計算スレッドの 25% をデータのプリフェッチに自動的に割り当てます。 説明 I/O が各モデルの全体的な計算パフォーマンスに与える影響は異なります。そのため、データのプリフェッチに使用するスレッド数を増やしても、モデル全体のトレーニング速度が必ずしも向上するとは限りません。
  capacity	いいえ	INT	0	プリフェッチするレコード数です。num_threads に指定した値が 1 より大きい場合、各スレッドは capacity/num_threads 件のデータレコードをプリフェッチします。この計算値は切り上げられます。capacity を 0 に設定した場合、組み込みリーダーは、テーブル内の先頭 N 件のレコードの平均値に基づいて、スレッドがプリフェッチできるデータの合計サイズを設定します。N のデフォルト値は 256 です。その結果、各スレッドがプリフェッチするデータのサイズは約 64 MB になります。

  説明

  MaxCompute テーブル内のフィールドのデータ型が DOUBLE の場合、TensorFlow はそのデータ型を np.float64 にマッピングします。

• レスポンス

  パイプラインを作成するための入力として使用できる Dataset オブジェクトを返します。

例

import os
import tensorflow as tf
import paiio

# 設定ファイルのパスを指定します。値は、設定ファイルが格納されているパスに置き換えてください。
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# 読み取り対象のテーブルを指定します。${your_projectname} は実際のプロジェクト名に、${table_name} はアクセスするテーブル名に置き換えてください。
table = ["odps://${your_projectname}/tables/${table_name}"]
# TableRecordDataset インターフェイスを指定して、テーブルの itemid 列と price 列を読み取ります。
dataset = paiio.data.TableRecordDataset(table,
                                       record_defaults=[0, 0.0],
                                       selected_cols="itemid,price",
                                       num_threads=1,
                                       capacity=10)
# エポック数を 2、バッチサイズを 3、プリフェッチするバッチ数を 100 に指定します。
dataset = dataset.repeat(2).batch(3).prefetch(100)

ids, prices = tf.compat.v1.data.make_one_shot_iterator(dataset).get_next()

with tf.compat.v1.Session() as sess:
    sess.run(tf.compat.v1.global_variables_initializer())
    sess.run(tf.compat.v1.local_variables_initializer())
    try:
        while True:
            batch_ids, batch_prices = sess.run([ids, prices])
            print("batch_ids:", batch_ids)
            print("batch_prices:", batch_prices)
    except tf.errors.OutOfRangeError:
        print("End of dataset")

TableReader

概要

TensorFlow に依存することなく、MaxCompute SDK の TableReader インタフェースを使用できます。これにより、MaxCompute テーブルにアクセスし、リアルタイムで I/O 結果を取得できます。

• Reader オブジェクトの作成とテーブルのオープン

  • 構文

  reader = paiio.python_io.TableReader(table,
                       selected_cols="",
                      excluded_cols="",
                       slice_id=0,
                      slice_count=1):

  • パラメータ

  パラメータ	必須	型	デフォルト値	説明
  テーブル	はい	STRING	なし	開きたい MaxCompute テーブルの名前です。テーブル名のフォーマット: odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/...
  selected_cols	いいえ	STRING	空文字列 ("")	選択する列。複数の列はカンマ (,) で区切ります。値は STRING 型である必要があります。このパラメーターがデフォルト値の場合、すべての列が読み込まれます。selected_cols と excluded_cols は、どちらか一方しか指定できません。
  excluded_cols	いいえ	STRING	空文字列 ("")	除外する列。複数の列はコンマ (,) で区切ります。値は STRING 型である必要があります。このパラメーターがデフォルト値の場合、すべての列が読み込まれます。selected_cols と excluded_cols は、どちらか一方のみ指定できます。
  slice_id	いいえ	INT	0	分散読み取りモードでのシャードの ID です。有効な値： [0, slice_count-1]。分散読み取りモードでは、テーブルは slice_count の値に基づいて複数のシャードに分割されます。システムは slice_id で指定されたシャードからデータを読み取ります。このパラメーターをデフォルト値の 0 に設定すると、すべてのテーブルレコードが読み取られます。
  スライス数	いいえ	INT	1	分散読み取りモードにおけるシャードの数です。多くの場合、この値はワーカー数と一致します。

  • 戻り値

    Reader オブジェクトが返されます。

• データレコードの読み取り

  • 構文

  reader.read(num_records=1)

  • パラメータ

    num_records は、順次読み込まれるデータレコードの数を指定します。 デフォルト値は 1 で、1 件のレコードが読み込まれることを指定します。 num_records パラメーターを未読のレコード数より大きい値に設定した場合、読み取られたすべてのレコードが返されます。 レコードが返されない場合、PAIIO.python_io.OutOfRangeException がスローされます。

  • 戻り値

    NumPy の n 次元配列 (またはレコード配列) が返されます。配列内の各要素は、テーブルレコードで構成されるタプルです。

• 特定のデータレコードからのデータの取得

  • 構文

  reader.seek(offset=0)

  • パラメータ

  offset は、データの取得を開始するデータレコードの ID を指定します。 レコード ID は 0 から始まります。 slice_id と slice_count を指定した場合、データは、対応するシャード内の offset で指定されたレコードの場所に基づいて取得されます。 offset がテーブル内のデータレコードの総数より大きい値に設定されている場合、範囲外の例外がスローされます。 前の seek 操作でテーブルに属さないレコードが返され、さらに seek 操作を続行すると、PAIIO.python_io.OutOfRangeException がスローされます。

  重要

  テーブル内の未読データレコードの数が、読み取り操作で指定したバッチサイズより少ない場合、未読データレコードが返され、例外はスローされません。別の seek 操作を続行すると、例外がスローされます。

  • 戻り値

    戻り値はありません。操作でエラーが発生した場合、システムは例外をスローします。

• テーブル内のデータレコードの総数の取得

  • 構文

  reader.get_row_count()

  • パラメータ

    なし

  • 戻り値

    テーブルのデータ件数が返されます。slice_idとslice_countを指定した場合、シャードのデータ件数が返されます。

• テーブルのスキーマの取得

  • 構文

  reader.get_schema()

  • パラメータ

    なし

  • 戻り値

  1 次元配列が返されます。配列内の各要素は、テーブル内の列のスキーマに対応します。次の表に、スキーマに含まれるパラメータを説明します。

  パラメータ	説明
  列名	列の名前です。
  typestr	MaxCompute のデータ型の名前です。
  pytype	typestr で指定された値に対応する Python のデータ型。

  次の表は、typestr および pytype によって指定できる値のマッピングを示します。

  typestr	pytype
  BIGINT	INT
  DOUBLE	FLOAT
  BOOLEAN	BOOL
  STRING	OBJECT
  DATETIME	INT
  MAP 説明 このデータ型は、PAI に組み込まれている TensorFlow では使用できません。	OBJECT

• テーブルを閉じる

  • 構文

  reader.close()

  • パラメータ

    なし

  • 戻り値

    戻り値はありません。操作でエラーが発生した場合、システムは例外をスローします。

例

たとえば、myproject という名前の MaxCompute プロジェクトに test という名前のテーブルを保存しているとします。次の表は、テーブルの内容の一部を示しています。

次のコードは、TableReader インターフェイスを使用して uid、name、および price 列のデータを読み取る例です。

import os
import paiio

# 設定ファイルのパスを指定します。値を設定ファイルが保存されているパスに置き換えます。
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# テーブルをオープンし、Reader オブジェクトを返します。${your_projectname} を MaxCompute プロジェクトの名前に、${table_name} をアクセスしたいテーブルの名前に置き換えます。
reader = paiio.python_io.TableReader("odps://myproject/tables/test", selected_cols="uid,name,price")

# テーブル内のデータレコードの総数を取得します。
total_records_num = reader.get_row_count() # 3 を返します

batch_size = 2
# テーブルを読み取り、[(uid, name, price)*2] フォーマットのレコード配列を返します。
records = reader.read(batch_size) # [(25, "Apple", 5.0), (38, "Pear", 4.5)] を返します。
records = reader.read(batch_size) # [(17, "Watermelon", 2.2)] を返します。
# 読み取りを続行すると、OutOfRange 例外がスローされます。

# reader を閉じます。
reader.close()

TableWriter

TableWriter は MaxCompute SDK をベースに実装されており、TensorFlow には依存しません。TableWriter を使用して、MaxCompute テーブルにデータを直接書き込めます。

概要

• Writer オブジェクトの作成とテーブルのオープン

  • 構文

    writer = paiio.python_io.TableWriter(table, slice_id=0)

    説明

    • このインターフェイスは、既存のデータを消去せずにテーブルへデータを書き込みます。

    • 新しく書き込まれたデータは、テーブルを閉じた後にのみ読み取れます。

  • パラメーター

    パラメーター	必須	タイプ	デフォルト値	説明
    table	はい	STRING	なし	オープンする MaxCompute テーブルの名前です。テーブル名のフォーマット： odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/...。
    slice_id	いいえ	INT	0	シャードの ID です。分散モードでは、書き込み競合を防ぐために、データは異なるシャードに書き込まれます。スタンドアロンモードでは、デフォルト値の 0 を使用します。分散モードでは、パラメータサーバー (PS) ノードを含む複数のワーカーが、slice_id で指定された同一のシャードにデータを書き込むと、書き込み操作が失敗します。

  • 戻り値

    Writer オブジェクトが返されます。

• データレコードの書き込み

  • 構文

    writer.write(values, indices)

  • パラメーター

    パラメーター	必須	タイプ	デフォルト値	説明
    values	はい	Array	なし	書き込み対象のデータレコードです。1 件以上のレコードを書き込めます。 1 件のレコードのみを書き込む場合は、values にはスカラーで構成されるタプル、リスト、または 1次元配列を設定します。values にリストまたは 1次元配列を設定した場合、レコード内のすべての列は同じデータ型になります。 複数のレコードを書き込む場合は、values にはリストまたは 1次元配列を設定します。値内の各要素は、タプル、リスト、または 1次元配列で表されるレコードに対応します。
    indices	はい	Array[INT]	なし	書き込み対象のデータレコードの列です。値には、整数のインデックスで構成されるタプル、リスト、または 1次元配列を指定します。indices で指定した値内の各数値は、レコードの列に対応します。たとえば、数値 i は列 i に対応します。列番号は 0 から始まります。

  • 戻り値

    値は返されません。書き込み操作中にエラーが発生した場合、システムは例外をスローして現在のプロセスを終了します。

• テーブルを閉じる

  • 構文

    writer.close()

    説明

    with 文では、close() メソッドを明示的に呼び出してテーブルを閉じる必要はありません。

  • パラメーター

    なし

  • 戻り値

    値は返されません。操作中にエラーが発生した場合、システムは例外をスローします。

  • サンプル結果

    with 文で TableWriter を使用します：

    with paiio.python_io.TableWriter(table) as writer:
      # 書き込み用の values を準備します。
        writer.write(values, indices)
        # このセクションの外側でテーブルが自動的に閉じられます。

使用例

import paiio
import os

# 設定ファイルのパスを指定します。値は、設定ファイルが保存されているパスに置き換えてください。
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# データを準備します。
values = [(25, "Apple", 5.0, False),
          (38, "Pear", 4.5, False),
          (17, "Watermelon", 2.2, False)]

# テーブルをオープンして Writer オブジェクトを取得します。${your_projectname} をご自身の MaxCompute プロジェクト名に、${table_name} をアクセス先のテーブル名に置き換えてください。
writer = paiio.python_io.TableWriter("odps://${your_projectname}/tables/${table_name}")

# テーブルの列 0 ～ 3 にデータを書き込みます。
writer.write(values, indices=[0, 1, 2, 3])

# Writer オブジェクトを使用してテーブルを閉じます。
writer.close()