MaxCompute:SQL

Parametreler

Dönüş değerleri

Hem execute_sql() hem de run_sql() yöntemleri Görev örnekleri bilgilerini döndürür.

Örnekler

Eşzamanlı ve eşzamansız çalıştırma karşılaştırması

# Synchronous: blocks until the statement finishes
o.execute_sql('select * from table_name')

# Asynchronous: returns immediately
instance = o.run_sql('select * from table_name')
print(instance.get_logview_address())  # Get the LogView URL
instance.wait_for_success()  # Block until the statement finishes

Çalışma zamanı ipuçlarını (hints) geçirin

o.execute_sql('select * from pyodps_iris', hints={'odps.stage.mapper.split.size': 16})

Genel ayarlar

Her sonraki execute_sql() çağrısına çalışma zamanı parametrelerini uygulamak için options.sql.settings değerini ayarlayın. Kullanılabilir parametreler için Bayrak parametreleri bölümüne bakın.

from odps import options
options.sql.settings = {'odps.stage.mapper.split.size': 16}
o.execute_sql('select * from pyodps_iris')  # Hints apply automatically

Yapılandırılmış veriler (SELECT)

SELECT sorguları yapılandırılmış kayıtlar döndürür. Her kayıt üzerinde yineleme yapmak için bir for döngüsü kullanın:

with o.execute_sql('select * from table_name').open_reader() as reader:
    for record in reader:
        print(record)

Yapılandırılmamış veriler (DESC ve diğer komutlar)

desc gibi komutlar ham metin döndürür. Bu verilere reader.raw üzerinden erişin:

with o.execute_sql('desc table_name').open_reader() as reader:
    print(reader.raw)

InstanceTunnel ve Result arayüzü

Varsayılan olarak open_reader() yöntemi Result arayüzünü kullanır; bu durum zaman aşımına veya döndürülen kayıt sayısının sınırlanmasına neden olabilir. Tam veri okuma işlemleri için InstanceTunnel özelliğini etkinleştirin:

Seçenek 1: Genel ayar

from odps import options
options.tunnel.use_instance_tunnel = True

Seçenek 2: Çağrı başına parametre

with o.execute_sql('select * from table_name').open_reader(tunnel=True) as reader:
    for record in reader:
        print(record)

PyODPS V0.7.7.1 ve sonraki sürümlerde open_reader() yöntemi bu şekilde tam verilerin okunmasını destekler.

• MaxCompute sürümünüz eskiyse veya InstanceTunnel bir hatayla karşılaşursa PyODPS bir uyarı oluşturur ve otomatik olarak Result arayüzüne geri döner. Nedeni belirlemek için uyarı mesajını kontrol edin.

• MaxCompute sürümünüz yalnızca Result arayüzünü destekliyorsa ve tüm sonuçlara ihtiyacınız varsa önce verileri başka bir tabloya yazın, ardından open_reader() ile bu tablodan okuma yapın. Bu yaklaşım, projenizin veri koruma mekanizmasına tabidir.

• InstanceTunnel hakkında daha fazla bilgi için InstanceTunnel bölümüne bakın.

Okuma sınırı modu

PyODPS varsayılan olarak bir örnekten okunan verileri sınırlamaz. Ancak proje sahibi veri koruması yapılandırdıysa ve options.tunnel.limit_instance_tunnel ayarı yapılmadıysa PyODPS kısıtlamaları algıladığında okuma sınırı modunu otomatik olarak etkinleştirir. Okuma sınırı modu çoğu durumda okumaları 10.000 satırla sınırlar.

Okuma sınırı modunu manuel olarak etkinleştirin (korumalı projeler):

# Option A: per-call
reader = instance.open_reader(limit=True)

# Option B: global
options.tunnel.limit_instance_tunnel = True

Tüm verileri okumak için okuma sınırı modunu devre dışı bırakın:

with o.execute_sql('select * from table_name').open_reader(tunnel=True, limit=False) as reader:
    for record in reader:
        print(record)

DataWorks gibi ortamlarda options.tunnel.limit_instance_tunnel değeri varsayılan olarak True olabilir. Tüm verileri okumak için open_reader() yöntemine hem tunnel=True hem de limit=False parametrelerini geçirin.

Korumalı projeler

Projeniz korunuyorsa ve tunnel=True, limit=False ayarı kısıtlamayı kaldırmıyorsa okuma izinlerini vermesi için proje sahibiyle iletişime geçin. Ayrıntılar için Proje veri koruması bölümüne bakın.

from odps.models import Schema

myfunc = '''\
from odps.udf import annotate
from odps.distcache import get_cache_file

@annotate('bigint->bigint')
class Example(object):
    def __init__(self):
        self.n = int(get_cache_file('test_alias_res1').read())

    def evaluate(self, arg):
        return arg + self.n
'''
res1 = o.create_resource('test_alias_res1', 'file', file_obj='1')
o.create_resource('test_alias.py', 'py', file_obj=myfunc)
o.create_function('test_alias_func',
                  class_type='test_alias.Example',
                  resources=['test_alias.py', 'test_alias_res1'])

table = o.create_table(
    'test_table',
    schema=Schema.from_lists(['size'], ['bigint']),
    if_not_exists=True
)

data = [[1, ], ]
# Write one row of data with value 1
o.write_table(table, 0, [table.new_record(it) for it in data])

with o.execute_sql(
    'select test_alias_func(size) from test_table').open_reader() as reader:
    print(reader[0][0])

res2 = o.create_resource('test_alias_res2', 'file', file_obj='2')
# Map res1 alias to res2 without modifying the UDF or resource
with o.execute_sql(
    'select test_alias_func(size) from test_table',
    aliases={'test_alias_res1': 'test_alias_res2'}).open_reader() as reader:
    print(reader[0][0])

from odps import options
options.biz_id = 'my_biz_id'
o.execute_sql('select * from pyodps_iris')