Platform For AI:PAIIO を使用した MaxCompute テーブルの読み書き

制限事項

• PAIIO は、TensorFlow 1.12、1.15、または 2.0 のイメージを使用する DLC ジョブでのみ使用できます。

• PAIIO はカスタムイメージをサポートしていません。

アカウント情報の設定

paiio モジュールを使用して MaxCompute テーブルの読み書きを行う前に、MaxCompute アカウントの AccessKey を設定する必要があります。PAI は設定ファイルから設定を読み取ります。このファイルはマウントされたファイルシステムに配置し、環境変数を使用してコードで参照できます。

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 のエンドポイントです。たとえば、中国 (上海) リージョンのエンドポイントは 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

API

TensorFlow コミュニティでは、従来のインターフェイスに代わるものとして、TensorFlow 1.2 以降の Dataset インターフェイスを使用して入力パイプラインを構築することを推奨します。複数の 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	読み取る列名をカンマ区切りで指定する文字列です。このパラメーターが None の場合、すべての列が読み取られます。このパラメーターは excluded_cols と併用できません。
  excluded_cols	いいえ	STRING	None	除外する列名のカンマ区切り文字列です。このパラメーターが None の場合、列は除外されません。このパラメーターは selected_cols とは併用できません。
  slice_id	いいえ	INT	0	分散読み取りでは、このパラメーターは読み取るデータシャードの 0 から始まるインデックスを指定します。システムは、テーブルを slice_count で指定されたシャード数に分割し、この slice_id に対応するシャードを読み取ります。 slice_id がデフォルトの 0 で、かつ slice_count が 1 の場合、テーブル全体が読み取られます。slice_count が 1 より大きい場合、最初のシャード (インデックス 0) のみが読み取られます。
  slice_count	いいえ	INT	1	分散読み取りの場合、このパラメーターはデータを分割するシャードの総数を指定します。この値は通常、ワーカー数に設定されます。デフォルト値の 1 は、テーブルがシャーディングされず、リーダーがテーブル全体を読み取ることを意味します。
  num_threads	いいえ	INT	0	リーダーが各テーブルのデータをプリフェッチするために使用する並列スレッドの数を指定します。これらのスレッドは、計算スレッドとは独立して動作します。値は 1 から 64 までの整数である必要があります。num_threads が 0 に設定されている場合、システムはプリフェッチスレッドの数を計算スレッドの数の 4 分の 1 に自動的に設定します。 説明 I/O の影響はモデルによって異なるため、プリフェッチスレッド数を増やしても、モデルのトレーニングが速くなることは保証されません。
  capacity	いいえ	INT	0	テーブルからプリフェッチする行の総数を指定します。num_threads が 1 より大きい場合、各スレッドのプリフェッチ容量は capacity/num_threads 行に切り上げられます。capacity が 0 に設定されている場合、組み込みの Reader は、テーブルの最初の 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"
# 読み取るテーブルを定義します。プレースホルダーを実際のプロジェクト名とテーブル名に置き換えてください。
table = ["odps://${your_projectname}/tables/${your_tablename}"]
# 'itemid' と 'price' 列を読み取るために TableRecordDataset を定義します。
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("データセットの終わり")

TableReader

API リファレンス

TableReader は MaxCompute SDK 上に構築されており、TensorFlow フレームワークとは独立して動作します。これにより、MaxCompute テーブルに直接アクセスし、I/O 結果をリアルタイムで取得できます。

• リーダーの作成とテーブルのオープン

  • 構文

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

  • パラメーター

  パラメーター	必須	タイプ	デフォルト	説明
  table	はい	STRING	N/A	オープンする MaxCompute テーブルの名前です。テーブル名は、odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/... の形式にする必要があります。
  selected_cols	いいえ	STRING	空の文字列 ("")	選択する列名をカンマ区切りで指定する文字列。空の文字列 ("") が指定された場合、すべての列が読み取られます。このパラメーターは excluded_cols と併用できません。
  excluded_cols	いいえ	STRING	空の文字列 ("")	除外する列名をカンマで区切った文字列です。空の文字列 ("") を指定すると、すべての列が読み込まれます。このパラメーターは selected_cols と併用できません。
  slice_id	いいえ	INT	0	分散読み取りシナリオでは、このパラメーターは現在のシャードのインデックスを指定します。値の範囲は [0, slice_count-1] です。分散モードで読み取る場合、システムは slice_count に基づいてテーブルを複数のシャードに分割し、slice_id で指定されたシャードを読み取ります。デフォルト値の 0 は、テーブルがシャード化されておらず、すべての行が読み取られることを示します。
  slice_count	いいえ	INT	1	分散読み取りシナリオでは、このパラメーターはシャードの総数を指定します。これは通常、ワーカー数になります。

  • 戻り値

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

• レコードの読み取り

  • 構文

  reader.read(num_records=1)

  • パラメーター

    num_records は、連続して読み取る行数を指定します。デフォルト値は 1 で、1 行を読み取ります。num_records が未読の行数を超えた場合、残りのすべての行が返されます。レコードが読み取られない場合、paiio.python_io.OutOfRangeException 例外がスローされます。

  • 戻り値

    NumPy ndarray (または recarray) が返されます。配列の各要素は、テーブルの行を表すタプルです。

• 特定の行へのシーク

  • 構文

  reader.seek(offset=0)

  • パラメーター

  offset は、シーク先の行を指定します (行インデックスは 0 から始まります)。 次の読み取り操作は、この行から開始されます。 slice_id と slice_count が設定されている場合、シークはシャード内の位置からの相対的なものになります。 offset がテーブルの総行数を超えた場合、OutOfRangeException がスローされます。 読み取り位置がすでにテーブルの末尾を超えている状態で再度シークしようとすると、同様に paiio.python_io.OutOfRangeException がスローされます。

  重要

  バッチの読み取り時に、残りの行数が batch_size 未満の場合、read 操作は例外をスローせずに残りの行を返します。この状態で別の seek 操作を試行すると、例外がスローされます。

  • 戻り値

    なし。エラーが発生した場合は例外がスローされます。

• 総行数の取得

  • 構文

  reader.get_row_count()

  • パラメーター

    なし

  • 戻り値

    テーブルの行数を返します。slice_id と slice_count が設定されている場合は、シャードのサイズを返します。

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

  • 構文

  reader.get_schema()

  • パラメーター

    なし

  • 戻り値

  1 次元の構造化 ndarray が返されます。各要素は MaxCompute テーブルから選択された列を記述し、次の 3 つのフィールドを含みます。

  パラメーター	説明
  colname	列名です。
  型文字列	MaxCompute のデータ型の名前です。
  pytype	typestr に対応する Python データ型です。

  次の表に、typestr と pytype のマッピングを示します。

  typestr	pytype
  BIGINT	INT
  DOUBLE	FLOAT
  BOOLEAN	BOOL
  STRING	OBJECT
  DATETIME	INT
  MAP 説明 PAI-TensorFlow は MAP データをサポートしていません。	OBJECT

• テーブルを閉じる

  • 構文

  reader.close()

  • パラメーター

    なし

  • 戻り値

    なし。エラーが発生した場合は例外がスローされます。

例

この例では、myproject プロジェクトの test という名前のテーブルで、次のデータを使用します：

次のコードは、TableReader を使用して itemid、name、および price カラムからデータを読み取る方法を示しています。

import os
import paiio

# 設定ファイルのパスを指定します。値を実際のパスに置き換えてください。
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# テーブルを開きます。プレースホルダーを実際のプロジェクト名とテーブル名に置き換えてください。
reader = paiio.python_io.TableReader("odps://${your_projectname}/tables/${your_tablename}", selected_cols="itemid,name,price")

# テーブルの総行数を取得します。
total_records_num = reader.get_row_count() # 3 を返す

batch_size = 2
# テーブルを読み取ります。戻り値は [(itemid, name, price)*2] 形式の recarray です。
records = reader.read(batch_size) # [(25, "Apple", 5.0), (38, "Pear", 4.5)] を返す
records = reader.read(batch_size) # [(17, "Watermelon", 2.2)] を返す
# 再度読み取ると OutOfRangeException がスローされます。

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

TableWriterの使用方法

TableWriter は MaxCompute SDK に基づいており、TensorFlow フレームワークに依存しないため、MaxCompute テーブルに直接データを書き込むことができます。

API

• ライターの作成とテーブルのオープン

  • 構文

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

    説明

    • この操作はテーブルにデータを追加するもので、既存のデータはクリアされません。

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

  • パラメーター

    パラメーター	必須	タイプ	デフォルト	説明
    table	はい	STRING	None	オープンする 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)

  • パラメーター

    パラメーター	必須	タイプ	デフォルト	説明
    値	はい	LIST、TUPLE、または ndarray	None	書き込むデータです。単一のレコードまたは複数のレコードとして指定します。 単一のレコードを書き込むには、values パラメーターに TUPLE、LIST、またはスカラーの 1D-ndarray を渡します。LIST または ndarray を渡す場合、レコード内のすべての列が同じデータ型である必要があります。 1 つ以上のレコードを書き込むには、リストまたは 1D-ndarray を values パラメーターに渡します。各要素は、1 つのレコードを表すタプル、リスト、または構造化 ndarray 要素である必要があります。
    インデックス	はい	LIST、TUPLE、または ndarray	None	書き込む列のインデックスを、整数のタプル、リスト、または 1次元 ndarray で指定します。indices の各インデックスは、0から始まる列番号です。

  • 戻り値

    正常に書き込まれたレコード数が返されます。操作が失敗した場合は、例外がスローされます。

• テーブルを閉じる

  • 構文

    writer.close()

    説明

    with 文を使用する場合、close() メソッドを明示的に呼び出す必要はありません。

  • パラメーター

    なし

  • 戻り値

    なし。エラーが発生した場合は、例外がスローされます。

  • 例

    次のコードは、with 文で TableWriter を使用する方法を示しています。

    with paiio.python_io.TableWriter(table) as writer:
        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 = paiio.python_io.TableWriter("odps://${your_projectname}/tables/${your_tablename}")

# テーブルの 0 から 3 列目にレコードを書き込みます。
records = writer.write(values, indices=[0, 1, 2, 3])

# ライターを閉じます。
writer.close()