Platform For AI:Usar o MaxCompute

Preparação: Configurar informações da conta

Antes de usar o PAIIO para ler ou gravar dados em tabelas do MaxCompute, configure as informações de AccessKey necessárias para acessar os recursos do MaxCompute. O PAI permite obter essas informações a partir de um arquivo de configuração. Para isso, armazene o arquivo em um sistema de arquivos e referencie as informações no seu código por meio de variáveis de ambiente.

1. Crie um arquivo de configuração com o seguinte conteúdo:

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

   Parâmetro	Descrição
   access_id	O AccessKey ID da conta Alibaba Cloud.
   access_key	O AccessKey secret da sua conta Alibaba Cloud.
   end_point	O endpoint do MaxCompute. Por exemplo, o endpoint da região China (Shanghai) é http://service.cn-shanghai.maxcompute.aliyun.com/api. Para mais informações, consulte Endpoints.

2. Especifique o caminho do arquivo de configuração no código usando a sintaxe abaixo:

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

   Substitua o parâmetro <your MaxCompute config file path> pelo caminho real do arquivo.

TableRecordDataset

Visão geral

O TensorFlow open source recomenda o uso do TensorFlow Datasets na versão 1.2 ou posterior para substituir as interfaces antigas de threading e filas na criação de fluxos de dados. Várias interfaces Dataset são combinadas para gerar dados destinados à computação, o que simplifica o código de entrada de dados.

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

• Parâmetros

  Parâmetro	Obrigatório	Tipo	Valor padrão	Descrição
  filenames	Sim	STRING	None	Nomes das tabelas a serem lidas. As tabelas devem compartilhar o mesmo schema. Formato do nome da tabela: odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/....
  record_defaults	Sim	LIST ou TUPLE	None	Tipo de dado da coluna a ser lida. Se a coluna estiver vazia, este parâmetro define o tipo de dado padrão. Caso haja incompatibilidade com o tipo da coluna lida ou a conversão automática falhe, o sistema lançará uma exceção. Valores válidos: FLOAT32, FLOAT64, INT32, INT64, BOOL e STRING. Para consultar os valores padrão do tipo INT64, utilize a sintaxe np.array(0, np.int64).
  selected_cols	Não	STRING	None	Colunas a serem selecionadas. Separe múltiplas colunas por vírgula (,). Se mantido o valor padrão None, todas as colunas serão lidas. É possível especificar apenas um dos parâmetros: selected_cols ou excluded_cols.
  excluded_cols	Não	STRING	None	Colunas a serem excluídas. Separe múltiplas colunas por vírgula (,). Se mantido o valor padrão None, todas as colunas serão lidas. É possível especificar apenas um dos parâmetros: selected_cols ou excluded_cols.
  slice_id	Não	INT	0	ID do shard no modo de leitura distribuída, iniciando em 0. Nesse modo, a tabela é dividida em vários shards conforme o valor de slice_count. O sistema lê os dados do shard indicado por slice_id. Se slice_id mantiver o valor padrão 0 e slice_count for 1, a tabela inteira será lida. Se slice_id for 0 e slice_count for maior que 1, apenas o primeiro shard (índice 0) será lido.
  slice_count	Não	INT	1	Quantidade de shards no modo de leitura distribuída. Geralmente corresponde ao número de workers. Se definido como 1 (padrão), a tabela inteira é lida sem fragmentação.
  num_threads	Não	INT	0	Número de threads ativadas pelo leitor interno de cada tabela para pré-busca de dados, independentes das threads de cálculo. Valores válidos: 1 a 64. Se num_threads for 0, o sistema aloca automaticamente 25% das threads de cálculo para pré-busca. Nota O impacto do I/O no desempenho geral varia conforme o modelo. Portanto, aumentar o número de threads de pré-busca não garante melhoria na velocidade de treinamento.
  capacity	Não	INT	0	Quantidade de registros pré-buscados. Se num_threads for maior que 1, cada thread fará a pré-busca de capacity/num_threads registros (valor arredondado para cima). Se capacity for 0, o leitor interno calcula o tamanho total de pré-busca com base na média dos primeiros N registros (padrão 256), resultando em aproximadamente 64 MB por thread.

  Nota

  Se os campos de uma tabela do MaxCompute forem do tipo DOUBLE, o TensorFlow os mapeará para np.float64.

• Respostas

  Retorna um objeto Dataset, utilizável como entrada para a criação de pipelines.

Exemplos

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

Visão geral

A interface TableReader, disponível no SDK do MaxCompute, dispensa a dependência do TensorFlow. Ela permite acessar tabelas do MaxCompute e obter resultados de I/O em tempo real.

• Criar um objeto Reader e abrir uma tabela

  • Sintaxe

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

  • Parâmetros

  Parâmetro	Obrigatório	Tipo	Valor padrão	Descrição
  table	Sim	STRING	None	Nome da tabela do MaxCompute a ser aberta. Formato: odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/...
  selected_cols	Não	STRING	String vazia ("")	Colunas a serem selecionadas. Separe múltiplas colunas por vírgula (,). O valor deve ser do tipo STRING. Se mantido o padrão, todas as colunas serão lidas. Especifique apenas selected_cols ou excluded_cols.
  excluded_cols	Não	STRING	String vazia ("")	Colunas a serem excluídas. Separe múltiplas colunas por vírgula (,). O valor deve ser do tipo STRING. Se mantido o padrão, todas as colunas serão lidas. Especifique apenas selected_cols ou excluded_cols.
  slice_id	Não	INT	0	ID do shard no modo de leitura distribuída. Valores válidos: [0, slice_count-1]. A tabela é dividida em shards conforme slice_count, e o sistema lê o shard definido por slice_id. O valor padrão 0 lê todos os registros da tabela.
  slice_count	Não	INT	1	Quantidade de shards no modo de leitura distribuída. Normalmente equivale ao número de workers.

  • Respostas

    Retorna um objeto Reader.

• Ler registros de dados

  • Sintaxe

  reader.read(num_records=1)

  • Parâmetros

    num_records define a quantidade de registros lidos sequencialmente. O padrão é 1 (registro único). Se num_records exceder o número de registros restantes não lidos, todos os registros disponíveis serão retornados. Se não houver registros, PAIIO.python_io.OutOfRangeException será lançado.

  • Respostas

    Retorna um array n-dimensional numpy (ou array de registros). Cada elemento é uma tupla contendo um registro da tabela.

• Obter dados a partir de um registro específico

  • Sintaxe

  reader.seek(offset=0)

  • Parâmetros

  offset indica o ID do registro inicial para obtenção de dados, começando em 0. Ao definir slice_id e slice_count, a busca considera a posição do registro dentro do shard correspondente. Se offset ultrapassar o total de registros da tabela, uma exceção de limite excedido será lançada. Caso uma operação seek anterior tenha retornado um registro inválido e outra busca seja tentada, PAIIO.python_io.OutOfRangeException ocorrerá.

  Importante

  Se os registros não lidos forem inferiores ao tamanho do lote solicitado, o sistema retorna apenas os disponíveis sem erro. Uma nova tentativa de seek, contudo, gerará uma exceção.

  • Respostas

    Nenhum valor é retornado. Em caso de erro na operação, o sistema lança uma exceção.

• Obter o total de registros da tabela

  • Sintaxe

  reader.get_row_count()

  • Parâmetros

    Nenhum

  • Respostas

    Retorna a contagem de registros. Se slice_id e slice_count estiverem definidos, retorna a contagem específica daquele shard.

• Obter o schema da tabela

  • Sintaxe

  reader.get_schema()

  • Parâmetros

    Nenhum

  • Respostas

  Retorna um array unidimensional onde cada elemento representa o schema de uma coluna. A tabela a seguir detalha os parâmetros do schema.

  Parâmetro	Descrição
  colname	Nome da coluna.
  typestr	Nome do tipo de dado MaxCompute.
  pytype	Tipo de dado Python correspondente ao valor de typestr.

  A tabela abaixo apresenta os mapeamentos entre os valores de typestr e pytype.

  typestr	pytype
  BIGINT	INT
  DOUBLE	FLOAT
  BOOLEAN	BOOL
  STRING	OBJECT
  DATETIME	INT
  MAP Nota Este tipo de dado não está disponível no TensorFlow integrado ao PAI.	OBJECT

• Fechar a tabela

  • Sintaxe

  reader.close()

  • Parâmetros

    Nenhum

  • Respostas

    Nenhum valor é retornado. Em caso de erro na operação, o sistema lança uma exceção.

Exemplos

Considere uma tabela chamada test no projeto do MaxCompute denominado myproject. A tabela a seguir exibe parte do seu conteúdo.

O exemplo abaixo ilustra o uso da interface TableReader para extrair dados das colunas uid, name e 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

Implementado com base no SDK do MaxCompute, o TableWriter independe do TensorFlow e permite a gravação direta de dados em tabelas do MaxCompute.

Visão geral

• Criar um objeto Writer e abrir uma tabela

  • Sintaxe

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

    Nota

    • Esta interface adiciona dados à tabela sem remover o conteúdo existente.

    • Os novos dados só ficam legíveis após o fechamento da tabela.

  • Parâmetros

    Parâmetro	Obrigatório	Tipo	Valor padrão	Descrição
    table	Sim	STRING	None	Nome da tabela do MaxCompute a ser aberta. Formato:odps://${your_projectname}/tables/${table_name}/${pt_1}/${pt_2}/....
    slice_id	Não	INT	0	ID do shard. No modo distribuído, os dados são escritos em shards distintos para evitar conflitos. No modo standalone, utilize o valor padrão 0. No modo distribuído, a gravação falhará se múltiplos workers (incluindo nós parameter server - PS) tentarem escrever no mesmo shard definido por slice_id.

  • Respostas

    Retorna um objeto Writer.

• Gravar registros de dados

  • Sintaxe

    writer.write(values, indices)

  • Parâmetros

    Parâmetro	Obrigatório	Tipo	Valor padrão	Descrição
    values	Sim	STRING	None	Registro(s) a ser(em) gravado(s). Para um único registro, defina values como uma tupla, lista ou array unidimensional de escalares. Se for lista ou array, todas as colunas devem ter o mesmo tipo. Para um ou mais registros, defina values como lista ou array unidimensional. Cada elemento representa um registro (tupla, lista ou array).
    indices	Sim	INT	None	Colunas alvo para gravação. Aceita tupla, lista ou array unidimensional de índices inteiros. Cada número em indices refere-se a uma coluna do registro (iniciando em 0).

  • Respostas

    Nenhum valor é retornado. Se ocorrer um erro durante a gravação, o sistema lança uma exceção e encerra o processo atual.

• Fechar a tabela

  • Sintaxe

    writer.close()

    Nota

    Ao utilizar a instrução WITH, não é necessário chamar explicitamente o método close() para fechar a tabela.

  • Parâmetros

    Nenhum

  • Respostas

    Nenhum valor é retornado. Em caso de erro na operação, o sistema lança uma exceção.

  • Resultados de exemplo

    Uso do TableWriter com a instrução 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.

Exemplos

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()