Platform For AI:Baca dan tulis tabel MaxCompute dengan PAIIO

Batasan

• PAIIO hanya dapat digunakan dengan pekerjaan DLC yang menggunakan image TensorFlow 1.12, 1.15, atau 2.0.

• PAIIO tidak mendukung custom image.

Konfigurasi informasi akun

Sebelum menggunakan modul PAIIO untuk membaca dari atau menulis ke tabel MaxCompute, Anda harus mengonfigurasi AccessKey akun MaxCompute Anda. PAI membaca konfigurasi dari sebuah file. Anda dapat menempatkan file tersebut di sistem file yang dimount dan mereferensikannya dalam kode menggunakan variabel lingkungan.

1. Buat file konfigurasi yang berisi konten berikut:

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

   Parameter	Deskripsi
   access_id	ID AccessKey Akun Alibaba Cloud Anda.
   access_key	Rahasia AccessKey Akun Alibaba Cloud Anda.
   end_point	Titik akhir untuk MaxCompute. Misalnya, titik akhir untuk wilayah China (Shanghai) adalah http://service.cn-shanghai.maxcompute.aliyun.com/api. Untuk informasi selengkapnya, lihat Endpoints.

2. Dalam kode Anda, tentukan path ke file konfigurasi sebagai berikut:

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

   Ganti <your MaxCompute config file path> dengan path aktual ke file konfigurasi Anda.

TableRecordDataset

API

Komunitas TensorFlow merekomendasikan penggunaan antarmuka Dataset di TensorFlow 1.2 dan versi setelahnya untuk membangun pipa input, menggantikan antarmuka thread dan queue lama. Anda dapat menggabungkan dan mentransformasi beberapa objek Dataset untuk menghasilkan data bagi komputasi, sehingga menyederhanakan kode input data.

• Definisi 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):

• Parameter

  Parameter	Wajib	Tipe	Default	Deskripsi
  filenames	Ya	STRING	-	Daftar tabel yang akan dibaca. Semua tabel harus memiliki skema yang sama. Nama tabel harus dalam format odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/....
  record_defaults	Ya	LIST atau TUPLE	-	Daftar atau tuple yang menentukan tipe data dan nilai default untuk setiap kolom yang akan dibaca. Metode ini melemparkan exception jika jumlah elemen tidak sesuai dengan jumlah kolom yang akan dibaca, atau jika tipe data tidak dapat dikonversi. Tipe data yang didukung meliputi FLOAT32, FLOAT64, INT32, INT64, BOOL, dan STRING. Untuk nilai default INT64, gunakan np.array(0, np.int64).
  selected_cols	Tidak	STRING	None	String nama kolom yang dipisahkan koma untuk dibaca. Jika parameter ini None, semua kolom dibaca. Parameter ini tidak dapat digunakan bersamaan dengan excluded_cols.
  excluded_cols	Tidak	STRING	None	String nama kolom yang dipisahkan koma untuk dikecualikan. Jika parameter ini None, tidak ada kolom yang dikecualikan. Parameter ini tidak dapat digunakan bersamaan dengan selected_cols.
  slice_id	Tidak	INT	0	Untuk pembacaan terdistribusi, parameter ini menentukan indeks berbasis 0 dari shard data yang akan dibaca. Sistem membagi tabel menjadi sejumlah shard yang ditentukan oleh slice_count dan membaca shard yang sesuai dengan slice_id ini. Jika slice_id adalah 0 (default) dan slice_count adalah 1, seluruh tabel dibaca. Jika slice_count lebih besar dari 1, hanya shard pertama (indeks 0) yang dibaca.
  slice_count	Tidak	INT	1	Untuk pembacaan terdistribusi, parameter ini menentukan jumlah total shard untuk membagi data. Nilai ini biasanya diatur sesuai jumlah worker. Nilai default 1 berarti tabel tidak di-shard dan reader membaca seluruh tabel.
  num_threads	Tidak	INT	0	Menentukan jumlah thread paralel yang digunakan reader untuk pra-ambil data untuk setiap tabel. Thread-thread ini beroperasi secara independen dari thread komputasi. Nilainya harus bilangan bulat antara 1 hingga 64. Jika num_threads diatur ke 0, sistem secara otomatis mengatur jumlah thread pra-ambil menjadi seperempat dari jumlah thread komputasi. Catatan Menambah jumlah thread pra-ambil tidak menjamin pelatihan model lebih cepat, karena dampak I/O bervariasi tergantung model.
  capacity	Tidak	INT	0	Menentukan jumlah total baris yang akan dipra-ambil dari tabel. Jika num_threads lebih besar dari 1, kapasitas pra-ambil untuk setiap thread adalah capacity/num_threads baris, dibulatkan ke atas. Jika capacity diatur ke 0, Reader bawaan secara otomatis mengonfigurasi kapasitas total pra-ambil berdasarkan ukuran rata-rata N baris pertama tabel, dengan N default 256. Hal ini memastikan bahwa jumlah data yang dipra-ambil untuk setiap thread kira-kira 64 MB.

  Catatan

  Jika suatu field dalam tabel MaxCompute bertipe data DOUBLE, Anda harus memetakannya ke np.float64 di TensorFlow.

• Nilai kembali

  Mengembalikan objek Dataset yang dapat digunakan untuk membangun pipa data.

Contoh

import os
import tensorflow as tf
import paiio

# Tentukan path ke file konfigurasi. Ganti ini dengan path file aktual.
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# Definisikan tabel yang akan dibaca. Ganti dengan nama proyek dan tabel aktual Anda.
table = ["odps://${your_projectname}/tables/${table_name}"]
# Definisikan TableRecordDataset untuk membaca kolom 'itemid' dan 'price'.
dataset = paiio.data.TableRecordDataset(table,
                                       record_defaults=[0, 0.0],
                                       selected_cols="itemid,price",
                                       num_threads=1,
                                       capacity=10)
# Atur 2 epoch, ukuran batch 3, dan pra-ambil 100 batch.
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

Referensi API

TableReader dibangun di atas SDK MaxCompute dan beroperasi secara independen dari framework TensorFlow. Antarmuka ini memungkinkan Anda mengakses tabel MaxCompute secara langsung dan mengambil hasil I/O secara real time.

• Buat reader dan buka tabel

  • Sintaks

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

  • Parameter

  Parameter	Wajib	Tipe	Default	Deskripsi
  table	Ya	STRING	N/A	Nama tabel MaxCompute yang akan dibuka. Nama tabel harus dalam format: odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/...
  selected_cols	Tidak	STRING	String kosong ("")	String nama kolom yang dipisahkan koma untuk dipilih. Jika string kosong ("") diberikan, semua kolom dibaca. Parameter ini tidak dapat digunakan bersamaan dengan excluded_cols.
  excluded_cols	Tidak	STRING	String kosong ("")	String nama kolom yang dipisahkan koma untuk dikecualikan. Jika string kosong ("") diberikan, semua kolom dibaca. Parameter ini tidak dapat digunakan bersamaan dengan selected_cols.
  slice_id	Tidak	INT	0	Dalam skenario pembacaan terdistribusi, parameter ini menentukan indeks shard saat ini. Nilainya dapat berkisar dari [0, slice_count-1]. Saat membaca dalam mode terdistribusi, sistem membagi tabel menjadi beberapa shard berdasarkan slice_count dan membaca shard yang ditentukan oleh slice_id. Nilai default 0 menunjukkan bahwa tabel tidak di-shard dan semua baris dibaca.
  slice_count	Tidak	INT	1	Dalam skenario pembacaan terdistribusi, parameter ini menentukan jumlah total shard, yang biasanya merupakan jumlah worker.

  • Nilai kembali

    Mengembalikan objek Reader.

• Baca catatan

  • Sintaks

  reader.read(num_records=1)

  • Parameter

    num_records menentukan jumlah baris yang akan dibaca secara berurutan. Nilai default adalah 1, yang membaca satu baris. Jika num_records melebihi jumlah baris yang belum dibaca, semua baris tersisa dikembalikan. Jika tidak ada catatan yang dibaca, exception paiio.python_io.OutOfRangeException dilemparkan.

  • Nilai kembali

    Mengembalikan ndarray NumPy (atau recarray). Setiap elemen dalam array adalah tuple yang merepresentasikan satu baris tabel.

• Lompat ke baris tertentu

  • Sintaks

  reader.seek(offset=0)

  • Parameter

  offset menentukan baris yang akan dilompati (pengindeksan baris dimulai dari 0). Operasi baca berikutnya dimulai dari baris ini. Jika slice_id dan slice_count dikonfigurasi, seek bersifat relatif terhadap posisi dalam shard. Jika offset melebihi jumlah total baris dalam tabel, exception OutOfRangeException dilemparkan. Jika posisi baca sudah melewati akhir tabel, mencoba seek lagi juga akan melemparkan exception paiio.python_io.OutOfRangeException.

  Penting

  Saat Anda membaca batch, jika jumlah baris tersisa kurang dari batch_size, operasi read mengembalikan baris tersisa tanpa melemparkan exception. Dalam kasus ini, mencoba operasi seek lagi akan melemparkan exception.

  • Nilai kembali

    None. Exception dilemparkan jika terjadi error.

• Dapatkan jumlah total baris

  • Sintaks

  reader.get_row_count()

  • Parameter

    None

  • Nilai kembali

    Mengembalikan jumlah baris dalam tabel. Jika slice_id dan slice_count dikonfigurasi, mengembalikan ukuran shard.

• Dapatkan skema tabel

  • Sintaks

  reader.get_schema()

  • Parameter

    None

  • Nilai kembali

  Mengembalikan ndarray terstruktur 1D. Setiap elemen menjelaskan kolom yang dipilih dari tabel MaxCompute dan berisi tiga field berikut.

  Parameter	Deskripsi
  colname	Nama kolom.
  typestr	Nama tipe data MaxCompute.
  pytype	Tipe data Python yang sesuai dengan typestr.

  Tabel berikut menjelaskan pemetaan antara typestr dan pytype.

  typestr	pytype
  BIGINT	INT
  DOUBLE	FLOAT
  BOOLEAN	BOOL
  STRING	OBJECT
  DATETIME	INT
  MAP Catatan PAI-TensorFlow tidak mendukung data MAP.	OBJECT

• Tutup tabel

  • Sintaks

  reader.close()

  • Parameter

    None

  • Nilai kembali

    None. Exception dilemparkan jika terjadi error.

Contoh

Contoh ini menggunakan tabel bernama test dalam proyek myproject dengan data berikut.

Kode berikut menunjukkan cara menggunakan TableReader untuk membaca data dari kolom uid, name, dan price.

import os
import paiio

# Tentukan path file konfigurasi. Ganti nilai ini dengan path aktual.
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# Buka tabel. Ganti myproject dan test dengan nama proyek dan tabel Anda.
reader = paiio.python_io.TableReader("odps://myproject/tables/test", selected_cols="uid,name,price")

# Dapatkan jumlah total baris dalam tabel.
total_records_num = reader.get_row_count() # return 3

batch_size = 2
# Baca tabel. Nilai kembali adalah recarray dalam format [(uid, name, price)*2].
records = reader.read(batch_size) # Mengembalikan [(25, "Apple", 5.0), (38, "Pear", 4.5)]
records = reader.read(batch_size) # Mengembalikan [(17, "Watermelon", 2.2)]
# Membaca lagi akan melemparkan OutOfRangeException.

# Tutup reader.
reader.close()

Penggunaan TableWriter

TableWriter didasarkan pada SDK MaxCompute dan tidak bergantung pada framework TensorFlow, memungkinkan Anda menulis data langsung ke tabel MaxCompute.

API

• Buat writer dan buka tabel

  • Sintaks

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

    Catatan

    • Operasi ini menambahkan data ke tabel dan tidak menghapus data yang sudah ada.

    • Anda hanya dapat membaca data yang baru ditulis setelah tabel ditutup.

  • Parameter

    Parameter	Wajib	Tipe	Default	Deskripsi
    table	Ya	STRING	None	Nama tabel MaxCompute yang akan dibuka. Nama harus dalam format berikut: odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/...
    slice_id	Tidak	INT	0	ID shard yang akan ditulis. Dalam mode terdistribusi, menulis ke shard berbeda mencegah konflik penulisan. Dalam mode standalone, Anda dapat menggunakan nilai default 0. Dalam mode terdistribusi, operasi penulisan akan gagal jika beberapa worker, termasuk node parameter server (PS), menulis ke shard yang sama menggunakan slice_id yang sama.

  • Nilai kembali

    Mengembalikan objek Writer.

• Tulis catatan

  • Sintaks

    writer.write(values, indices)

  • Parameter

    Parameter	Wajib	Tipe	Default	Deskripsi
    values	Ya	STRING	None	Data yang akan ditulis, ditentukan sebagai satu catatan atau beberapa catatan: Untuk menulis satu catatan, berikan TUPLE, LIST, atau ndarray 1D skalar ke parameter values. Jika Anda memberikan LIST atau ndarray, semua kolom dalam catatan harus memiliki tipe data yang sama. Untuk menulis satu atau lebih catatan, berikan LIST atau ndarray 1D ke parameter values. Setiap elemen harus berupa TUPLE, LIST, atau elemen ndarray terstruktur yang merepresentasikan satu catatan.
    indices	Ya	INT	None	Indeks kolom yang akan ditulis. Ini dapat berupa TUPLE, LIST, atau ndarray 1D bilangan bulat. Setiap indeks dalam indices adalah nomor kolom berbasis nol.

  • Nilai kembali

    Jumlah catatan yang berhasil ditulis. Jika operasi gagal, exception dilemparkan.

• Tutup tabel

  • Sintaks

    writer.close()

    Catatan

    Anda tidak perlu memanggil metode close() secara eksplisit saat menggunakan pernyataan with.

  • Parameter

    None

  • Nilai kembali

    None. Jika terjadi error, exception dilemparkan.

  • Contoh

    Kode berikut menunjukkan cara menggunakan TableWriter dengan pernyataan with.

    with paiio.python_io.TableWriter(table) as writer:
      # Siapkan nilai untuk ditulis.
        writer.write(values, indices)
        # Writer ditutup secara otomatis saat blok 'with' keluar.

Contoh

import paiio
import os

# Tentukan path file konfigurasi. Ganti nilai ini dengan path aktual.
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# Siapkan data.
values = [(25, "Apple", 5.0, False),
          (38, "Pear", 4.5, False),
          (17, "Watermelon", 2.2, False)]

# Buka tabel untuk mendapatkan objek writer. Ganti nama proyek dan tabel dengan nilai aktual Anda.
writer = paiio.python_io.TableWriter("odps://project/tables/test")

# Tulis catatan ke kolom 0 hingga 3 tabel.
records = writer.write(values, indices=[0, 1, 2, 3])

# Tutup writer.
writer.close()