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/${table_name}"]
# '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("End of dataset")

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 操作を試みると例外がスローされます。

  • 戻り値

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

• 総行数の取得

  • 構文

  reader.get_row_count()

  • パラメーター

    None

  • 戻り値

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

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

  • 構文

  reader.get_schema()

  • パラメーター

    None

  • 戻り値

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

  パラメーター	説明
  colname	列名。
  typestr	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()

  • パラメーター

    None

  • 戻り値

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

例

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

次のコードは、TableReader を使用して uid、name、price の各列からデータを読み取る方法を示しています。

import os
import paiio

# 設定ファイルのパスを指定します。値を実際のパスに置き換えてください。
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# テーブルを開きます。myproject と test を実際のプロジェクト名とテーブル名に置き換えてください。
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] 形式の 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)

  • パラメーター

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

  • 戻り値

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

• テーブルを閉じる

  • 構文

    writer.close()

    説明

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

  • パラメーター

    None

  • 戻り値

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

  • 例

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

    with paiio.python_io.TableWriter(table) as writer:
      # 書き込み用の値を準備します。
        writer.write(values, indices)
        # 'with' ブロックを抜けると、ライターは自動的に閉じられます。

例

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://project/tables/test")

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

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