Platform For AI:Utiliser MaxCompute

Préparation : configurer les informations de compte

Avant d'utiliser PAIIO pour lire ou écrire des données dans les tables MaxCompute, vous devez configurer les informations AccessKey utilisées pour accéder aux ressources MaxCompute. PAI vous permet d'obtenir ces informations à partir d'un fichier de configuration. Pour ce faire, stockez le fichier de configuration dans un système de fichiers et référencez les informations dans votre code à l'aide de variables d'environnement.

1. Créez un fichier de configuration contenant le contenu suivant :

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

   Paramètre	Description
   access_id	L'AccessKey ID du compte Alibaba Cloud.
   access_key	L'AccessKey secret de votre compte Alibaba Cloud.
   end_point	L'endpoint de MaxCompute. Par exemple, l'endpoint de la région Chine (Shanghai) est http://service.cn-shanghai.maxcompute.aliyun.com/api. Pour plus d'informations, consultez la rubrique Endpoints.

2. Utilisez la syntaxe suivante pour spécifier le chemin du fichier de configuration dans le code :

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

   Définissez le paramètre <your MaxCompute config file path> avec le chemin du fichier.

TableRecordDataset

Présentation

La version open source de TensorFlow recommande d'utiliser TensorFlow Datasets avec TensorFlow 1.2 ou une version ultérieure pour remplacer les interfaces de threading et de mise en file d'attente originales lors de la création de flux de données. Plusieurs interfaces Dataset sont combinées pour générer des données destinées au calcul, ce qui simplifie le code d'entrée des données.

• Syntaxe (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):

• Paramètres

  Parameter	Required	Type	Default value	Description
  filenames	Yes	STRING	None	Les noms des tables que vous souhaitez lire. Les tables doivent utiliser le même schéma. Format du nom de table : odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/....
  record_defaults	Yes	LIST or TUPLE	None	Le type de données de la colonne que vous souhaitez lire. Si la colonne est vide, ce paramètre indique le type de données par défaut. Si le type de données est incohérent avec celui de la colonne lue ou s'il ne peut pas être converti automatiquement, le système lève une exception. Valeurs valides : FLOAT32, FLOAT64, INT32, INT64, BOOL et STRING. Pour obtenir des informations sur les valeurs par défaut du type de données INT64, utilisez la syntaxe np.array(0, np.int64) pour la requête.
  selected_cols	No	STRING	None	La colonne que vous souhaitez sélectionner. Séparez plusieurs colonnes par des virgules (,). Si vous définissez ce paramètre sur la valeur par défaut None, toutes les colonnes sont lues. Vous ne pouvez spécifier qu'un seul des paramètres selected_cols et excluded_cols.
  excluded_cols	No	STRING	None	La colonne que vous souhaitez exclure. Séparez plusieurs colonnes par des virgules (,). Si vous définissez ce paramètre sur la valeur par défaut None, toutes les colonnes sont lues. Vous ne pouvez spécifier qu'un seul des paramètres selected_cols et excluded_cols.
  slice_id	No	INT	0	L'ID du shard en mode de lecture distribuée. L'ID du shard commence à 0. En mode de lecture distribuée, la table est divisée en plusieurs shards en fonction de la valeur du paramètre slice_count. Le système lit les données du shard spécifié par le paramètre slice_id. Si slice_id est défini sur la valeur par défaut 0 et slice_count sur 1, la table entière est lue. Si slice_id est défini sur la valeur par défaut 0 et slice_count sur une valeur supérieure à 1, le premier shard (indice 0) est lu.
  slice_count	No	INT	1	Le nombre de shards en mode de lecture distribuée. Dans la plupart des cas, cette valeur correspond au nombre de workers. Si vous définissez ce paramètre sur la valeur par défaut 1, la table entière est lue sans fragmentation.
  num_threads	No	INT	0	Le nombre de threads activés par le lecteur intégré de chaque table pour la prélecture des données. Ces threads sont indépendants des threads de calcul. Valeurs valides : 1 à 64. Si num_threads est défini sur 0, le système attribue automatiquement 25 % des threads de calcul à la prélecture des données. Remarque Les opérations d'E/S ont un impact différent sur les performances de calcul globales de chaque modèle. Par conséquent, l'augmentation du nombre de threads utilisés pour la prélecture des données n'améliore pas nécessairement la vitesse d'entraînement du modèle global.
  capacity	No	INT	0	Le nombre d'enregistrements préchargés. Si la valeur spécifiée par num_threads est supérieure à 1, chaque thread précharge capacity/num_threads enregistrements de données. La valeur du paramètre est arrondie à l'entier supérieur. Si capacity est défini sur 0, le lecteur intégré configure la taille totale des données que les threads peuvent précharger en fonction de la valeur moyenne des N premiers enregistrements de la table. La valeur par défaut de N est 256. Par conséquent, la taille des données préchargées par chaque thread est d'environ 64 Mo.

  Remarque

  Si le type de données des champs d'une table MaxCompute est DOUBLE, TensorFlow mappe ce type de données sur np.float64.

• Réponses

  Un objet Dataset est renvoyé et peut être utilisé comme entrée pour créer des pipelines.

Exemples

import os
import tensorflow as tf
import paiio

# Specify the path of the configuration file. Replace the value with the path where the configuration file is stored. 
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# Specify the tables that you want to read. Replace ${your_projectname} with the name of the MaxCompute project and ${table_name} with the name of the table that you want to access. 
table = ["odps://${your_projectname}/tables/${table_name}"]
# Specify the TableRecordDataset interface to read the itemid and price columns of the table. 
dataset = paiio.data.TableRecordDataset(table,
                                       record_defaults=[0, 0.0],
                                       selected_cols="itemid,price",
                                       num_threads=1,
                                       capacity=10)
# Specify epoch 2, batch size 3, and prefetch 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

Présentation

Vous pouvez utiliser l'interface TableReader du SDK MaxCompute sans dépendre de TensorFlow. Cette approche vous permet d'accéder aux tables MaxCompute et d'obtenir des résultats d'E/S en temps réel.

• Créer un objet Reader et ouvrir une table

  • Syntaxe

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

  • Paramètres

  Paramètre	Obligatoire	Type	Valeur par défaut	Description
  table	Oui	STRING	Aucune	Nom de la table MaxCompute à ouvrir. Format du nom de table : odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/...
  selected_cols	Non	STRING	Chaîne vide ("")	Colonne à sélectionner. Séparez plusieurs colonnes par des virgules (,). La valeur doit être de type STRING. Si ce paramètre conserve sa valeur par défaut, toutes les colonnes sont lues. Vous ne pouvez spécifier qu'un seul des paramètres selected_cols et excluded_cols.
  excluded_cols	Non	STRING	Chaîne vide ("")	Colonne à exclure. Séparez plusieurs colonnes par des virgules (,). La valeur doit être de type STRING. Si ce paramètre conserve sa valeur par défaut, toutes les colonnes sont lues. Vous ne pouvez spécifier qu'un seul des paramètres selected_cols et excluded_cols.
  slice_id	Non	INT	0	ID du shard en mode de lecture distribuée. Valeurs valides : [0, slice_count-1]. En mode de lecture distribuée, la table est divisée en plusieurs shards selon la valeur de slice_count. Le système lit les données depuis le shard indiqué par slice_id. Si vous conservez la valeur par défaut 0, tous les enregistrements de la table sont lus.
  slice_count	Non	INT	1	Nombre de shards en mode de lecture distribuée. Généralement, cette valeur correspond au nombre de workers.

  • Réponses

    Un objet Reader est renvoyé.

• Lire des enregistrements de données

  • Syntaxe

  reader.read(num_records=1)

  • Paramètres

    num_records indique le nombre d'enregistrements de données lus séquentiellement. La valeur par défaut est 1, ce qui signifie qu'un seul enregistrement est lu. Si vous définissez le paramètre num_records sur une valeur supérieure au nombre d'enregistrements non lus, tous les enregistrements disponibles sont renvoyés. Si aucun enregistrement n'est renvoyé, l'exception PAIIO.python_io.OutOfRangeException est levée.

  • Réponses

    Un tableau numpy multidimensionnel (ou record array) est renvoyé. Chaque élément du tableau est un tuple représentant un enregistrement de table.

• Obtenir des données à partir d'un enregistrement spécifique

  • Syntaxe

  reader.seek(offset=0)

  • Paramètres

  offset spécifie l'ID de l'enregistrement de données à partir duquel vous souhaitez obtenir les données. L'ID d'enregistrement commence à 0. Si vous spécifiez slice_id et slice_count, les données sont obtenues en fonction de l'emplacement de l'enregistrement indiqué par offset dans le shard correspondant. Si offset est défini sur une valeur supérieure au nombre total d'enregistrements de données dans la table, une exception de dépassement de plage est levée. Si l'opération seek précédente renvoie un enregistrement qui n'appartient pas à la table et que vous effectuez une autre opération seek, l'exception PAIIO.python_io.OutOfRangeException est levée.

  Important

  Si le nombre d'enregistrements de données non lus dans la table est inférieur à la taille de lot spécifiée pour une opération de lecture, le nombre d'enregistrements non lus est renvoyé et aucune exception n'est levée. Si vous effectuez une autre opération seek, une exception est levée.

  • Réponses

    Aucune valeur n'est renvoyée. En cas d'erreur lors d'une opération, le système lève une exception.

• Obtenir le nombre total d'enregistrements de données dans la table

  • Syntaxe

  reader.get_row_count()

  • Paramètres

    Aucun

  • Réponses

    Le nombre d'enregistrements de données dans la table est renvoyé. Si vous spécifiez slice_id et slice_count, le nombre d'enregistrements de données dans le shard est renvoyé.

• Obtenir le schéma de la table

  • Syntaxe

  reader.get_schema()

  • Paramètres

    Aucun

  • Réponses

  Un tableau unidimensionnel est renvoyé. Chaque élément du tableau correspond au schéma d'une colonne de la table. Le tableau suivant décrit les paramètres contenus dans un schéma.

  Paramètre	Description
  colname	Nom de la colonne.
  typestr	Nom du type de données MaxCompute.
  pytype	Type de données Python correspondant à la valeur spécifiée par typestr.

  Le tableau suivant décrit les mappages entre les valeurs pouvant être spécifiées par typestr et pytype.

  typestr	pytype
  BIGINT	INT
  DOUBLE	FLOAT
  BOOLEAN	BOOL
  STRING	OBJECT
  DATETIME	INT
  MAP Remarque Ce type de données n'est pas disponible pour TensorFlow intégré à PAI.	OBJECT

• Fermer la table

  • Syntaxe

  reader.close()

  • Paramètres

    Aucun

  • Réponses

    Aucune valeur n'est renvoyée. En cas d'erreur lors d'une opération, le système lève une exception.

Exemples

Par exemple, vous stockez une table nommée test dans votre projet MaxCompute nommé myproject. Le tableau suivant répertorie une partie du contenu de la table.

Le code suivant montre comment utiliser l'interface TableReader pour lire les données contenues dans les colonnes uid, name et price.

import os
import paiio

# Specify the path of the configuration file. Replace the value with the path where the configuration file is stored. 
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# Open a table and return a Reader object. Replace ${your_projectname} with the name of the MaxCompute project and ${table_name} with the name of the table that you want to access. 
reader = paiio.python_io.TableReader("odps://myproject/tables/test", selected_cols="uid,name,price")

# Obtain the total number of data records in the table. 
total_records_num = reader.get_row_count() # return 3

batch_size = 2
# Read the table and return a record array in the [(uid, name, price)*2] format. 
records = reader.read(batch_size) # Return [(25, "Apple", 5.0), (38, "Pear", 4.5)].
records = reader.read(batch_size) # Return [(17, "Watermelon", 2.2)].
# If you continue to read, an OutOfRange exception is thrown. 

# Close the reader.
reader.close()

TableWriter

TableWriter est implémenté à partir du SDK MaxCompute et ne dépend pas de TensorFlow. Vous pouvez utiliser TableWriter pour écrire des données directement dans une table MaxCompute.

Présentation

• Créez un objet Writer et ouvrez une table

  • Syntaxe

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

    Remarque

    • Cette interface écrit des données dans une table sans effacer les données existantes.

    • Les données nouvellement écrites ne peuvent être lues qu'après la fermeture de la table.

  • Paramètres

    Parameter	Required	Type	Default value	Description
    table	Yes	STRING	None	Nom de la table MaxCompute que vous souhaitez ouvrir. Format du nom de la table :odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/....
    slice_id	No	INT	0	ID du shard. En mode distribué, les données sont écrites dans différents shards pour éviter les conflits d'écriture. En mode autonome, utilisez la valeur par défaut 0. En mode distribué, l'opération d'écriture échoue si plusieurs workers, y compris les nœuds Parameter Server (PS), écrivent des données dans le même shard spécifié par slice_id.

  • Réponses

    Un objet Writer est renvoyé.

• Écrivez des enregistrements de données

  • Syntaxe

    writer.write(values, indices)

  • Paramètres

    Parameter	Required	Type	Default value	Description
    values	Yes	STRING	None	Enregistrement de données que vous souhaitez écrire. Vous pouvez écrire un ou plusieurs enregistrements. Pour écrire un seul enregistrement, définissez values sur un tuple, une liste ou un tableau unidimensionnel composé de scalaires. Si values est défini sur une liste ou un tableau unidimensionnel, toutes les colonnes de l'enregistrement sont du même type de données. Pour écrire un ou plusieurs enregistrements, définissez values sur une liste ou un tableau unidimensionnel. Chaque élément de la valeur correspond à un enregistrement qui est un tuple, une liste ou un tableau unidimensionnel.
    indices	Yes	INT	None	Colonnes de l'enregistrement de données que vous souhaitez écrire. La valeur peut être un tuple, une liste ou un tableau unidimensionnel composé d'index entiers. Chaque nombre de la valeur spécifiée par indices correspond à une colonne de l'enregistrement. Par exemple, le nombre i correspond à la colonne i. La numérotation des colonnes commence à 0.

  • Réponses

    Aucune valeur n'est renvoyée. Si une erreur se produit lors de l'opération d'écriture, le système lève une exception et quitte le processus actuel.

• Fermez la table

  • Syntaxe

    writer.close()

    Remarque

    Dans l'instruction WITH, vous n'avez pas besoin d'appeler explicitement la méthode close() pour fermer une table.

  • Paramètres

    Aucun

  • Réponses

    Aucune valeur n'est renvoyée. Si une erreur se produit lors d'une opération, le système lève une exception.

  • Exemples de résultats

    Utilisez TableWriter dans l'instruction WITH :

    with paiio.python_io.TableWriter(table) as writer:
      # Prepare values for writing.
        writer.write(values, incides)
        # Table would be closed automatically outside this section.

Exemples

import paiio
import os

# Specify the path of the configuration file. Replace the value with the path where the configuration file is stored. 
os.environ['ODPS_CONFIG_FILE_PATH'] = "/mnt/data/odps_config.ini"
# Prepare data. 
values = [(25, "Apple", 5.0, False),
          (38, "Pear", 4.5, False),
          (17, "Watermelon", 2.2, False)]

# Open a table and return a Writer object. Replace ${your_projectname} with the name of the MaxCompute project and ${table_name} with the name of the table that you want to access. 
writer = paiio.python_io.TableWriter("odps://project/tables/test")

# Write data to columns 0 to 3 of the table. 
records = writer.write(values, indices=[0, 1, 2, 3])

# Use the Writer object to close the table. 
writer.close()