MaxCompute:Tabelas

from odps.models import Schema

# Define columns
schema = Schema.from_lists(['id', 'name'], ['bigint', 'string'])

# Create the table
table = o.create_table('my_table', schema, if_not_exists=True)

# Write records
o.write_table('my_table', [[1, 'Alice'], [2, 'Bob']])

# Read records
for record in o.read_table('my_table'):
    print(record)

for table in o.list_tables():
    print(table)

for table in o.list_tables(prefix="table_prefix"):
    print(table.name)

for table in o.list_tables(extended=True):
    print(table.name, table.creation_time)

# Valid types: managed_table, external_table, virtual_view, materialized_view
managed_tables = list(o.list_tables(type="managed_table"))
external_tables = list(o.list_tables(type="external_table"))
virtual_views = list(o.list_tables(type="virtual_view"))
materialized_views = list(o.list_tables(type="materialized_view"))

print(o.exist_table('pyodps_iris'))
# Returns True if the table exists

t = o.get_table('pyodps_iris')

print(t.schema)

odps.Schema {
  sepallength           double      # Sepal length (cm)
  sepalwidth            double      # Sepal width (cm)
  petallength           double      # Petal length (cm)
  petalwidth            double      # Petal width (cm)
  name                  string      # Type
}

# All columns
print(t.schema.columns)

# A specific column
print(t.schema['sepallength'])

# Column comment
print(t.schema['sepallength'].comment)

print(t.lifecycle)        # Table lifecycle
print(t.creation_time)    # Creation time
print(t.is_virtual_view)  # Whether the table is a view
print(t.size)             # Table size in bytes
print(t.comment)          # Table comment

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 objetos Column e Partition do módulo 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 schema:

# 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 método Schema.from_lists() é mais simples, mas não oferece suporte a comentários em colunas:

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 schema

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

Ativar tipos de dados estendidos

Por padrão, apenas estes tipos de dados têm suporte: 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>')

from odps.models import Schema

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

# Fetch the latest table metadata from the server
table.reload()

write_table()

Use write_table() para gravações simples e únicas, quando todos os registros já estiverem prontos. Esse método adiciona dados à tabela e gerencia 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 método write_table() sempre adiciona dados. Para substituir dados existentes:

• Tabelas sem partição: 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 em 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 multiprocessos

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 confirma a operação 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 iterar sobre todos os registros de 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 toda a tabela:

t = o.get_table('my_new_table')

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

open_reader()

Use open_reader() quando precisar de acesso fatiado ou da contagem de registros antes da leitura. Esse método expõe um atributo count e oferece suporte a 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 um bloco with:

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

# Delete only if the table exists
o.delete_table('my_table_name', if_exists=True)

# Or call drop() on a table object
t.drop()

table = o.get_table('my_table_name')
df = table.to_df()

Verificar se uma tabela é particionada

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

Iterar sobre 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, 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)

Criar uma partição

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

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

odps.Schema {
  c_int_a                 bigint
  c_string_a              string
  c_bool_a                boolean
  c_datetime_a            datetime
  c_array_a               array<string>
  c_map_a                 map<bigint,string>
  c_struct_a              struct<a:bigint,b:string>
}

import datetime

t = o.get_table('mytable')  # o is the MaxCompute entry object

# Create a record with initial values
# The number of values must match the number of fields in the schema
r = t.new_record([1024, 'val1', False, datetime.datetime.now(), None, None])

# Create an empty record
r2 = t.new_record()

# Set values by index
r2[0] = 1024

# Set values by field name
r2['c_string_a'] = 'val1'

# Set values by attribute
r2.c_string_a = 'val1'

# Set ARRAY value
r2.c_array_a = ['val1', 'val2']

# Set MAP value
r2.c_map_a = {1: 'val1'}

# Set STRUCT value (PyODPS 0.11.5 and later)
r2.c_struct_a = (1, 'val1')             # tuple
r2.c_struct_a = {"a": 1, "b": 'val1'}   # dict

# Get values
print(r[0])                          # By index
print(r['c_string_a'])               # By field name
print(r.c_string_a)                  # By attribute
print(r[0: 3])                       # Slice
print(r[0, 2, 3])                    # Multiple indices
print(r['c_int_a', 'c_double_a'])    # Multiple field names

Mapeamentos de tipos de dados

Tratamento de STRING

Por padrão, 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 de 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 timestamps UNIX para armazenamento.

Para usar 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, STRUCT era mapeado para dict. A partir do PyODPS 0.11.5, STRUCT é mapeado para namedtuple por padrão.

Para restaurar o comportamento anterior com dict:

options.struct_as_dict = True

Em ambientes DataWorks, struct_as_dict tem como padrão False 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])

Fazer download de 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)