MaxCompute:Tables

Acessar tabelas entre projetos

Passe o parâmetro project para obter uma tabela de outro projeto:

t = o.get_table('table_name', project='other_project')

Método 1: Objetos Column e Partition

Use os objetos Column e Partition do odps.models quando precisar de controle total sobre as definições de colunas, incluindo comentários:

from odps.models import Schema, Column, Partition

columns = [
    Column(name='num', type='bigint', comment='the column'),
    Column(name='num2', type='double', comment='the column2'),
]
partitions = [Partition(name='pt', type='string', comment='the partition')]
schema = Schema(columns=columns, partitions=partitions)

Acesse as propriedades do esquema:

# All columns including partition columns
print(schema.columns)

# Partition columns only
print(schema.partitions)

# Non-partition column names
print(schema.names)

# Non-partition column types
print(schema.types)

Método 2: Schema.from_lists()

O Schema.from_lists() é mais simples, mas não suporta comentários de coluna:

from odps.models import Schema

schema = Schema.from_lists(
    ['num', 'num2'],           # Column names
    ['bigint', 'double'],      # Column types
    ['pt'],                    # Partition names
    ['string']                 # Partition types
)
print(schema.columns)

A partir de um objeto de esquema

from odps.models import Schema

schema = Schema.from_lists(['num', 'num2'], ['bigint', 'double'], ['pt'], ['string'])

# Create a table
table = o.create_table('my_new_table', schema)

# Skip creation if the table already exists
table = o.create_table('my_new_table', schema, if_not_exists=True)

# Set the lifecycle (days before auto-deletion)
table = o.create_table('my_new_table', schema, lifecycle=7)

A partir de definições de colunas como strings

# Partitioned table (columns, partition columns)
table = o.create_table('my_new_table', ('num bigint, num2 double', 'pt string'), if_not_exists=True)

# Non-partitioned table
table = o.create_table('my_new_table02', 'num bigint, num2 double', if_not_exists=True)

Ative tipos de dados estendidos

Por padrão, há suporte apenas para os seguintes tipos de dados: BIGINT, DOUBLE, DECIMAL, STRING, DATETIME, BOOLEAN, MAP e ARRAY.

Para usar tipos estendidos como TINYINT e STRUCT, ative a extensão de tipos de dados do MaxCompute V2.0:

from odps import options

options.sql.use_odps2_extension = True
table = o.create_table('my_new_table', 'cat smallint, content struct<title:varchar(100), body:string>')

write_table()

Use write_table() para gravações simples e pontuais, quando todos os registros já estiverem prontos. O método acrescenta dados à tabela e lida com partições em uma única chamada.

records = [
    [111, 1.0],
    [222, 2.0],
    [333, 3.0],
    [444, 4.0]
]

# Write to a partition; create the partition if it does not exist
o.write_table('my_new_table', records, partition='pt=test', create_partition=True)

O write_table() sempre acrescenta dados. Para substituir os dados existentes:

• Tabelas não particionadas: Chame table.truncate().

• Tabelas particionadas: Exclua e recrie a partição.

open_writer()

Use open_writer() para gravações em streaming ou para gravar registros incrementalmente dentro de uma sessão gerenciada.

t = o.get_table('my_new_table')

with t.open_writer(partition='pt=test02', create_partition=True) as writer:
    records = [
        [1, 1.0],
        [2, 2.0],
        [3, 3.0],
        [4, 4.0]
    ]
    writer.write(records)  # Accepts any iterable

Grave em partições de vários níveis:

t = o.get_table('test_table')

with t.open_writer(partition='pt1=test1,pt2=test2') as writer:
    records = [
        t.new_record([111, 'aaa', True]),
        t.new_record([222, 'bbb', False]),
        t.new_record([333, 'ccc', True]),
        t.new_record([444, 'Chinese', False])
    ]
    writer.write(records)

Gravação paralela com múltiplos processos

Vários processos podem gravar na mesma tabela simultaneamente ao compartilhar um ID de sessão e gravar em blocos separados. Cada bloco corresponde a um arquivo no servidor. O processo principal executa o commit após a conclusão de todos os workers.

import random
from multiprocessing import Pool
from odps.tunnel import TableTunnel

def write_records(tunnel, table, session_id, block_id):
    # Reuse the existing session
    local_session = tunnel.create_upload_session(table.name, upload_id=session_id)
    # Write to this process's block
    with local_session.open_record_writer(block_id) as writer:
        for i in range(5):
            record = table.new_record([random.randint(1, 100), random.random()])
            writer.write(record)

if __name__ == '__main__':
    N_WORKERS = 3

    table = o.create_table('my_new_table', 'num bigint, num2 double', if_not_exists=True)
    tunnel = TableTunnel(o)
    upload_session = tunnel.create_upload_session(table.name)

    # Share the session ID across processes
    session_id = upload_session.id

    pool = Pool(processes=N_WORKERS)
    futures = []
    block_ids = []
    for i in range(N_WORKERS):
        futures.append(pool.apply_async(write_records, (tunnel, table, session_id, i)))
        block_ids.append(i)
    [f.get() for f in futures]

    # Commit all blocks
    upload_session.commit(block_ids)

read_table()

Use read_table() para percorrer todos os registros em uma tabela ou partição:

for record in o.read_table('my_new_table', partition='pt=test'):
    print(record)

head()

Visualize até 10.000 registros sem varrer a tabela inteira:

t = o.get_table('my_new_table')

for record in t.head(3):
    print(record)

open_reader()

Use open_reader() quando precisar de acesso por fatias ou da contagem de registros antes da leitura. Ele expõe o atributo count e suporta fatiamento por índice:

t = o.get_table('my_new_table')

with t.open_reader(partition='pt=test') as reader:
    count = reader.count
    for record in reader[5:10]:  # Read a slice of records
        print(record)

Sem o bloco with:

reader = t.open_reader(partition='pt=test')
count = reader.count
for record in reader[5:10]:
    print(record)

Verificar se uma tabela é particionada

table = o.get_table('my_new_table')
if table.schema.partitions:
    print('Table %s is partitioned.' % table.name)

Iterar pelas partições

table = o.get_table('my_new_table')

# All partitions
for partition in table.partitions:
    print(partition.name)

# Sub-partitions under pt=test
for partition in table.iterate_partitions(spec='pt=test'):
    print(partition.name)

# Partitions matching a condition (PyODPS 0.11.3 and later)
for partition in table.iterate_partitions(spec='dt>20230119'):
    print(partition.name)

A partir do PyODPS 0.11.3, o iterate_partitions() aceita expressões lógicas como dt>20230119.

Verificar se uma partição existe

table = o.get_table('my_new_table')
table.exist_partition('pt=test,sub=2015')

Obter informações da partição

table = o.get_table('my_new_table')
partition = table.get_partition('pt=test')
print(partition.creation_time)
print(partition.size)

Crie uma partição

t = o.get_table('my_new_table')
t.create_partition('pt=test', if_not_exists=True)

Exclua uma partição

t = o.get_table('my_new_table')
t.delete_partition('pt=test', if_exists=True)

# Or call drop() on a partition object
partition.drop()

Mapeamentos de tipos de dados

Tratamento de STRING

Por padrão, o tipo STRING é mapeado para strings Unicode (str no Python 3, unicode no Python 2). Para armazenar dados binários em uma coluna STRING, defina options.tunnel.string_as_binary = True.

Comportamento do fuso horário

O PyODPS usa o fuso horário local por padrão. O MaxCompute não armazena valores de fuso horário — ele converte valores datetime em carimbos de data e hora UNIX para armazenamento.

Para usar o UTC:

options.local_timezone = False

Para usar um fuso horário específico:

options.local_timezone = 'Asia/Shanghai'

DECIMAL no Python 2

Quando o pacote cdecimal está instalado, o PyODPS usa cdecimal.Decimal em vez de decimal.Decimal no Python 2.

Comportamento do tipo STRUCT

Antes do PyODPS 0.11.5, o tipo STRUCT era mapeado para dict. A partir do PyODPS 0.11.5, o tipo STRUCT é mapeado para namedtuple por padrão.

Para restaurar o comportamento anterior de dict:

options.struct_as_dict = True

Em ambientes do DataWorks, struct_as_dict tem False como padrão para compatibilidade histórica. O PyODPS 0.11.5 e versões posteriores aceitam tanto dict quanto tuple para valores STRUCT. Versões anteriores aceitam apenas dict.

Fazer upload de dados

from odps.tunnel import TableTunnel

table = o.get_table('my_table')

tunnel = TableTunnel(o)
upload_session = tunnel.create_upload_session(table.name, partition_spec='pt=test')

with upload_session.open_record_writer(0) as writer:
    record = table.new_record()
    record[0] = 'test1'
    record[1] = 'id1'
    writer.write(record)

    record = table.new_record(['test2', 'id2'])
    writer.write(record)

# Commit outside the with block. Committing before all data is written causes an error.
upload_session.commit([0])

Baixe dados

from odps.tunnel import TableTunnel

tunnel = TableTunnel(o)
download_session = tunnel.create_download_session('my_table', partition_spec='pt=test')

with download_session.open_record_reader(0, download_session.count) as reader:
    for record in reader:
        print(record)