MaxCompute:Tables

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

Projeler arasında tablolara erişim

Başka bir projeden bir tablo almak için project parametresini kullanın:

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

Yöntem 1: Sütun ve Bölüm nesneleri

Yorumlar dahil olmak üzere sütun tanımları üzerinde tam kontrole ihtiyacınız olduğunda odps.models içindeki Column ve Partition nesnelerini kullanın:

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)

Şema özelliklerine erişin:

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

Yöntem 2: Schema.from_lists()

Schema.from_lists() daha basittir ancak sütun yorumlarını desteklemez:

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)

Şema nesnesinden

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)

Dize olarak sütun tanımlarından

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

Genişletilmiş veri türlerini etkinleştirme

Varsayılan olarak yalnızca şu veri türleri desteklenir: BIGINT, DOUBLE, DECIMAL, STRING, DATETIME, BOOLEAN, MAP ve ARRAY.

TINYINT ve STRUCT gibi genişletilmiş türleri kullanmak için MaxCompute V2.0 veri türü uzantısını etkinleştirin:

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

Tüm kayıtlar hazırken tek seferlik basit yazma işlemleri için write_table() yöntemini kullanın. Bu yöntem tabloya veri ekler ve bölümleri tek bir çağrıda işler.

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)

write_table() her zaman veri ekler. Mevcut verileri değiştirmek için:

• Bölümlenmemiş tablolar: table.truncate() yöntemini çağırın.

• Bölümlenmiş tablolar: Bölümü silin ve yeniden oluşturun.

open_writer()

Akış yazmaları veya yönetilen bir oturum içinde kayıtları kademeli olarak yazmak için open_writer() yöntemini kullanın.

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

Çok düzeyli bölümlere yazın:

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)

Çok işlemli paralel yazma

Birden fazla işlem, bir oturum kimliğini paylaşarak ve ayrı bloklara yazarak aynı tabloya eşzamanlı olarak yazabilir. Her blok sunucuda bir dosyaya karşılık gelir. Ana işlem, tüm çalışanlar tamamlandıktan sonra commit işlemini gerçekleştirir.

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

Bir tablodaki veya bölümdeki tüm kayıtlar üzerinde yineleme yapmak için read_table() yöntemini kullanın:

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

head()

Tam tablo taraması yapmadan en fazla 10.000 kaydı önizleyin:

t = o.get_table('my_new_table')

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

open_reader()

Okumadan önce dilimlenmiş erişime veya kayıt sayısına ihtiyaç duyduğunuzda open_reader() yöntemini kullanın. Bu yöntem bir count niteliği sunar ve dizin dilimlemeyi destekler:

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)

with bloğu olmadan:

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

Bir tablonun bölümlenip bölünmediğini kontrol etme

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

Bölümler üzerinde yineleme yapma

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)

PyODPS 0.11.3'ten başlayarak iterate_partitions() yöntemi dt>20230119 gibi mantıksal ifadeleri kabul eder.

Bir bölümün varlığını kontrol etme

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

Bölüm bilgilerini alma

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

Bölüm oluşturma

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

Bölüm silme

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

Veri türü eşlemeleri

STRING işleme

Varsayılan olarak STRING, Unicode dizelerine eşlenir (Python 3'te str, Python 2'de unicode). Bir STRING sütununda ikili veri depolamak için options.tunnel.string_as_binary = True ayarını yapın.

Saat dilimi davranışı

PyODPS varsayılan olarak yerel saat dilimini kullanır. MaxCompute saat dilimi değerlerini saklamaz; datetime değerlerini depolama için UNIX zaman damgalarına dönüştürür.

UTC kullanmak için:

options.local_timezone = False

Belirli bir saat dilimi kullanmak için:

options.local_timezone = 'Asia/Shanghai'

Python 2'de DECIMAL

cdecimal paketi yüklendiğinde PyODPS, Python 2'de decimal.Decimal yerine cdecimal.Decimal kullanır.

STRUCT türü davranışı

PyODPS 0.11.5 öncesinde STRUCT, dict'e eşlenirdi. PyODPS 0.11.5'ten itibaren STRUCT varsayılan olarak namedtuple'a eşlenir.

Önceki dict davranışını geri yüklemek için:

options.struct_as_dict = True

DataWorks ortamlarında, struct_as_dict geçmiş uyumluluğu nedeniyle varsayılan olarak False değerine sahiptir. PyODPS 0.11.5 ve sonraki sürümler STRUCT değerleri için hem dict hem de tuple kabul eder. Daha eski sürümler yalnızca dict kabul eder.

Veri yükleme

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

Veri indirme

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)