MaxCompute：Tabel

Informasi latar belakang

Siapkan lingkungan runtime

Operasi dasar

Operasi pada tabel dalam proyek

• Kueri semua tabel dalam proyek.

  Anda dapat menggunakan metode o.list_tables() untuk menanyakan semua tabel dalam proyek.

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

  Anda dapat menentukan parameter prefix untuk menanyakan tabel dengan awalan tertentu.

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

  Metode ini hanya menampilkan nama tabel. Properti tabel lainnya, seperti table_schema dan creation_time, tidak ditampilkan. Jika Anda ingin mendapatkan properti tersebut, permintaan tambahan diperlukan, dan waktu yang lebih lama akan dikonsumsi. Di PyODPS 0.11.5 dan versi terbaru, Anda dapat menambahkan konfigurasi extended=True ke metode list_tables untuk mendapatkan properti tabel tambahan.

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

  Jika Anda ingin menanyakan tabel berdasarkan jenis, tentukan parameter type. Contoh:

  managed_tables=list (o.list_tables(type="managed_table"))  # Kueri tabel internal.
  external_tables=list (o.list_tables(type="external_table"))  # Kueri tabel eksternal.
  virtual_views=list (o.list_tables(type="virtual_view"))  # Kueri tampilan.
  materialized_views=list (o.list_tables(type="materialized_view"))  # Kueri tampilan materialisasi.

• Periksa apakah sebuah tabel ada.

  Anda dapat menggunakan metode o.exist_table() untuk memeriksa apakah sebuah tabel ada.

  print(o.exist_table('pyodps_iris'))
  # Jika True dikembalikan, tabel pyodps_iris ada.

• Peroleh informasi tentang sebuah tabel.

  Peroleh informasi tentang sebuah tabel dengan memanggil metode o.get_table() dari objek entri MaxCompute.

  • Peroleh informasi skema sebuah tabel.

    t = o.get_table('pyodps_iris')
    print(t.schema) # Peroleh informasi skema tabel pyodps_iris.

    Contoh respons:

    odps.Schema {
      sepallength           double      # Panjang sepal (cm)
      sepalwidth            double      # Lebar sepal (cm)
      petallength           double      # Panjang petal (cm)
      petalwidth            double      # Lebar petal (cm)
      name                  string      # Jenis
    }

  • Kueri detail kolom dalam sebuah tabel.

    t = o.get_table('pyodps_iris')
    print(t.schema.columns) # Kueri detail kolom dalam skema tabel pyodps_iris.

    Contoh respons:

    [<column sepallength, type double>,
     <column sepalwidth, type double>,
     <column petallength, type double>,
     <column petalwidth, type double>,
     <column name, type string>]

  • Kueri detail kolom dalam sebuah tabel.

    t = o.get_table('pyodps_iris')
    print(t.schema['sepallength']) # Dapatkan informasi tentang kolom sepallength tabel pyodps_iris.

    Contoh respons:

    <column sepallength, type double>

  • Peroleh komentar kolom dalam sebuah tabel.

    t = o.get_table('pyodps_iris')
    print(t.schema['sepallength'].comment) # Peroleh detail kolom sepallength dalam tabel pyodps_iris.

    Contoh respons:

    Panjang sepal (cm)

  • Peroleh siklus hidup sebuah tabel.

    t = o.get_table('pyodps_iris')
    print(t.lifecycle) # Peroleh siklus hidup tabel pyodps_iris.

    Contoh respons:

    -1

  • Peroleh waktu pembuatan sebuah tabel.

    t = o.get_table('pyodps_iris')
    print(t.creation_time) # Peroleh waktu pembuatan tabel pyodps_iris.

  • Periksa apakah sebuah tabel adalah tampilan virtual.

    t = o.get_table('pyodps_iris')
    print(t.is_virtual_view) # Periksa apakah tabel pyodps_iris adalah tampilan virtual. Jika False dikembalikan, tabel pyodps_iris bukan tampilan virtual.

  Seperti contoh sebelumnya, Anda juga dapat menggunakan metode t.size dan t.comment untuk mendapatkan ukuran tabel dan komentar tabel.

  Operasi pada tabel lintas proyek

  Anda dapat memperoleh informasi tentang tabel dari proyek lain dengan menentukan parameter project.

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

  Dalam kode sebelumnya, atur other_project ke nama proyek dari mana Anda ingin memperoleh informasi tentang tabel, dan atur table_name ke nama tabel yang informasinya ingin Anda peroleh.

Buat skema tabel

Buat tabel

Anda dapat memanggil metode o.create_table() untuk membuat tabel dengan menggunakan skema tabel atau dengan menentukan nama dan tipe data kolom. Saat membuat tabel, Anda harus memastikan bahwa tipe data kolom dalam tabel valid.

Gunakan skema tabel untuk membuat tabel

# Buat skema tabel.
from odps.models import Schema
schema = Schema.from_lists(['num', 'num2'], ['bigint', 'double'], ['pt'], ['string'])

# Buat tabel menggunakan skema yang telah dibuat.
table = o.create_table('my_new_table', schema)

# Buat tabel hanya jika tidak ada tabel dengan nama yang sama.
table = o.create_table('my_new_table', schema, if_not_exists=True)

# Konfigurasikan siklus hidup tabel.
table = o.create_table('my_new_table', schema, lifecycle=7)

Anda dapat memanggil metode print(o.exist_table('my_new_table')) untuk memeriksa apakah tabel berhasil dibuat. Jika True dikembalikan, tabel berhasil dibuat.

Buat tabel dengan menentukan nama dan tipe data kolom yang akan disertakan dalam tabel

# Buat tabel terpartisi bernama my_new_table dengan kolom umum dan kolom kunci partisi yang ditentukan.
table = o.create_table('my_new_table', ('num bigint, num2 double', 'pt string'), if_not_exists=True)

# Buat tabel non-partisi bernama my_new_table02.
table = o.create_table('my_new_table02', 'num bigint, num2 double', if_not_exists=True)

Anda dapat memanggil metode print(o.exist_table('my_new_table')) untuk memeriksa apakah tabel berhasil dibuat. Jika True dikembalikan, tabel berhasil dibuat.

Buat tabel dengan menentukan nama dan tipe data kolom yang akan disertakan dalam tabel: tipe data baru dalam edisi tipe data 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>')

Sinkronkan pembaruan tabel

Setelah program lain memperbarui tabel, seperti skema tabel, Anda dapat memanggil metode reload() untuk menyinkronkan pembaruan.

# Ubah skema tabel.
from odps.models import Schema
schema = Schema.from_lists(['num', 'num2'], ['bigint', 'double'], ['pt'], ['string'])

# Panggil metode reload() untuk menyinkronkan pembaruan.
table = o.create_table('my_new_table', schema)
table.reload()

Gunakan tipe Record

Tipe Record adalah struktur data yang digunakan untuk mewakili satu baris data dalam tabel MaxCompute. Ini digunakan oleh antarmuka Table.open_reader atau Table.open_writer untuk membaca atau menulis data. Ini juga digunakan oleh antarmuka Tunnel TableDownloadSession.open_record_reader atau TableUploadSession.open_record_writer. Anda dapat memanggil metode new_record pada objek tabel untuk membuat instance Record.

Struktur tabel sampel:

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>
}

Kode berikut menunjukkan cara membuat instance Record untuk tabel dan melakukan operasi terkait:

import datetime
t = o.get_table('mytable')  # o adalah objek entri MaxCompute.
r = t.new_record([1024, 'val1', False, datetime.datetime.now(), None, None])  # Jumlah nilai harus sama dengan jumlah bidang dalam skema tabel.
r2 = t.new_record()  # Anda dapat meninggalkan nilai kosong selama inisialisasi.
r2[0] = 1024  # Konfigurasikan nilai berdasarkan offset.
r2['c_string_a'] = 'val1'  # Konfigurasikan nilai berdasarkan nama bidang.
r2.c_string_a = 'val1'  # Konfigurasikan nilai berdasarkan atribut.
r2.c_array_a = ['val1', 'val2']  # Konfigurasikan nilai tipe ARRAY.
r2.c_map_a = {1: 'val1'}  # Konfigurasikan nilai tipe MAP.
r2.c_struct_a = (1, 'val1')  # Gunakan tuple untuk mengonfigurasi nilai tipe STRUCT di PyODPS 0.11.5 atau lebih baru.
r2.c_struct_a = {"a": 1, "b": 'val1'}  # Anda juga dapat menggunakan dict untuk mengonfigurasi nilai tipe STRUCT.

print(r[0])  # Dapatkan nilai pada posisi 0.
print(r['c_string_a'])  # Dapatkan nilai berdasarkan bidang.
print(r.c_string_a)  # Dapatkan nilai berdasarkan atribut.
print(r[0: 3])  # Lakukan operasi slicing.
print(r[0, 2, 3])  # Dapatkan nilai di beberapa posisi.
print(r['c_int_a', 'c_double_a'])  # Dapatkan nilai berdasarkan beberapa bidang.

Tabel berikut mencantumkan pemetaan antara tipe data MaxCompute dan tipe data Python dalam Record.

Catatan:

• Secara default, data tipe STRING di PyODPS sesuai dengan string Unicode, yang direpresentasikan sebagai str di Python 3 dan unicode di Python 2. Dalam skenario di mana data BINARY disimpan sebagai data tipe STRING, Anda harus mengonfigurasi options.tunnel.string_as_binary = True; untuk menghindari masalah penyandian potensial.

• PyODPS menggunakan zona waktu lokal secara default. Jika Anda menggunakan zona waktu UTC, Anda harus mengonfigurasi options.local_timezone = False;. Jika Anda menggunakan zona waktu lain, Anda harus menyetel parameter ini ke zona waktu tertentu, seperti Asia/Shanghai. MaxCompute tidak menyimpan nilai zona waktu. Oleh karena itu, saat data ditulis, informasi waktu dikonversi menjadi timestamp UNIX untuk penyimpanan.

• Untuk Python 2, kelas cdecimal.Decimal digunakan saat paket cdecimal diinstal.

• Di versi PyODPS sebelum 0.11.5, tipe STRUCT di MaxCompute sesuai dengan tipe dict di Python. Di PyODPS 0.11.5 dan yang lebih baru, tipe STRUCT di MaxCompute sesuai dengan tipe namedtuple di Python. Jika Anda ingin menggunakan perilaku lama, Anda harus mengonfigurasi options.struct_as_dict = True;. Di lingkungan DataWorks, untuk menjaga kompatibilitas historis, parameter ini disetel ke False secara default. Saat mengonfigurasi nilai tipe STRUCT untuk bidang dalam Record, PyODPS 0.11.5 dan yang lebih baru mendukung tipe dict dan tuple, sedangkan versi PyODPS yang lebih lama hanya mendukung tipe dict.

• Untuk informasi lebih lanjut tentang cara mengonfigurasi parameter, lihat Konfigurasi.

Tulis data ke tabel

• Panggil metode write_table() dari objek entri MaxCompute untuk menulis data ke tabel.

  Penting

  Jika tabel terpartisi tidak berisi partisi tempat Anda ingin menulis data, Anda dapat mengonfigurasi parameter create_partition untuk membuat partisi.

  records = [[111, 1.0],                 # Daftar dapat ditentukan.
            [222, 2.0],
            [333, 3.0],
            [444, 4.0]]
  o.write_table (my_new_table, records, partition='pt=test, create_partition=True) # Buat partisi bernama test dan tulis data ke partisi tersebut.

  Catatan

  • Setiap kali Anda memanggil metode write_table(), MaxCompute menghasilkan file di server. Operasi ini memakan waktu. Selain itu, jika terlalu banyak file dihasilkan, efisiensi kueri selanjutnya akan terpengaruh. Kami sarankan Anda menulis beberapa rekaman sekaligus atau memberikan objek generator jika Anda menggunakan metode write_table().

  • Jika Anda memanggil metode write_table() untuk menulis data ke tabel, data baru akan ditambahkan ke data yang ada. PyODPS tidak menyediakan opsi untuk menimpa data yang ada. Anda harus secara manual menghapus data yang ingin Anda timpa. Untuk tabel non-partisi, Anda harus memanggil metode table.truncate() untuk menghapus data. Untuk tabel terpartisi, Anda harus menghapus partisi dan kemudian membuat partisi lagi.

• Panggil metode open_writer() untuk menulis data ke tabel.

  t = o.get_table('my_new_table')
  with t.open_writer(partition='pt=test02', create_partition=True) as writer:  # Buat partisi bernama test02 dan tulis data ke partisi tersebut.
      records = [[1, 1.0], # Daftar dapat ditentukan.
                [2, 2.0],
                [3, 3.0],
                [4, 4.0]]
      writer.write(records)  # Rekaman dapat berupa objek iterable.

  Kode berikut menunjukkan cara menulis data ke tabel terpartisi multi-level:

  t = o.get_table('test_table')
  with t.open_writer(partition='pt1=test1,pt2=test2') as writer:  # Tulis data dalam mode partisi multi-level.
      records = [t.new_record([111, 'aaa', True]),   # Objek Record dapat digunakan.
                 t.new_record([222, 'bbb', False]),
                 t.new_record([333, 'ccc', True]),
                 t.new_record([444, 'Chinese', False])]
      writer.write(records)

• Gunakan proses multiproses untuk menulis data ke tabel secara bersamaan.

  Jika beberapa proses menulis data ke tabel secara bersamaan, semua proses menggunakan ID sesi yang sama tetapi menulis data ke blok yang berbeda. Setiap blok sesuai dengan file di server. Setelah semua proses selesai menulis data, proses utama mengirimkan data.

  import random
  from multiprocessing import Pool
  from odps.tunnel import TableTunnel
  def write_records(tunnel, table, session_id, block_id):
      # Buat sesi dengan ID sesi yang ditentukan.
      local_session = tunnel.create_upload_session(table.name, upload_id=session_id)
      # Buat penulis dengan ID blok yang ditentukan.
      with local_session.open_record_writer(block_id) as writer:
          for i in range(5):
              # Hasilkan data dan tulis data ke blok yang benar.
              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)
  
      # Semua proses menggunakan ID sesi yang sama.
      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]
  
      # Kirim data di semua blok.
      upload_session.commit(block_ids)

Peroleh data tabel

Hapus tabel

o.delete_table('my_table_name', if_exists=True)  # Hapus tabel hanya jika tabel tersebut ada.
t.drop() # Panggil metode drop() untuk menghapus tabel jika tabel tersebut ada.

Konversikan tabel menjadi DataFrame

PyODPS menyediakan kerangka kerja DataFrame, yang memungkinkan Anda dengan mudah menanyakan dan mengelola data MaxCompute. Untuk informasi lebih lanjut, lihat DataFrame (tidak direkomendasikan). Anda dapat memanggil metode to_df() untuk mengonversi tabel menjadi DataFrame.

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

Kelola partisi tabel

• Periksa apakah sebuah tabel dipartisi.

  table = o.get_table('my_new_table')
  if table.schema.partitions:
      print('Tabel %s dipartisi.' % table.name)

• Iterasi semua partisi dalam tabel.

  table = o.get_table('my_new_table')
  for partition in table.partitions:  # Iterasi semua partisi.
      print(partition.name) # Langkah iterasi. Pada langkah ini, nama partisi ditampilkan.
  for partition in table.iterate_partitions(spec='pt=test'):  # Iterasi partisi level-2 dalam partisi bernama test.
      print(partition.name) # Langkah iterasi. Pada langkah ini, nama partisi ditampilkan.
  for partition in table.iterate_partitions(spec='dt>20230119'):  # Iterasi partisi level-2 dalam partisi yang memenuhi kondisi dt>20230119.
      print(partition.name)  # Langkah iterasi. Pada langkah ini, nama partisi ditampilkan.

  Penting

  Di PyODPS 0.11.3 dan yang lebih baru, Anda dapat menentukan ekspresi logis untuk iterate_partitions, seperti dt>20230119 dalam contoh sebelumnya.

• Periksa apakah sebuah partisi ada.

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

• Peroleh informasi tentang sebuah partisi.

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

• Buat partisi.

  t = o.get_table('my_new_table')
  t.create_partition('pt=test', if_not_exists=True)  # Buat partisi hanya jika tidak ada partisi dengan nama yang sama.

• Hapus partisi yang ada.

  t = o.get_table('my_new_table')
  t.delete_partition('pt=test', if_exists=True)  # Setel parameter if_exists ke True. Ini memastikan bahwa partisi dihapus hanya jika partisi tersebut ada.
  partition.drop()  # Panggil metode drop() untuk menghapus partisi jika partisi tersebut ada.

Unggah dan unduh data menggunakan MaxCompute Tunnel

MaxCompute Tunnel adalah terowongan data MaxCompute. Anda dapat menggunakan Tunnel untuk mengunggah data ke atau mengunduh data dari MaxCompute.

• Contoh unggah data

  from odps.tunnel import TableTunnel
  
  table = o.get_table('my_table')
  
  tunnel = TableTunnel(odps)
  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)
  
  # Anda harus mengeksekusi pernyataan berikut di luar blok kode WITH. Jika Anda mengeksekusi pernyataan berikut sebelum data ditulis, kesalahan akan dilaporkan.
  upload_session.commit([0])

• Contoh unduh data

  from odps.tunnel import TableTunnel
  
  tunnel = TableTunnel(odps)
  download_session = tunnel.create_download_session('my_table', partition_spec='pt=test')
  # Proses setiap rekaman.
  with download_session.open_record_reader(0, download_session.count) as reader:
      for record in reader:
          print(record)  # Langkah iterasi. Pada langkah ini, rekaman ditampilkan.