Python SDK Kullanım Örnekleri: SQL İfadeleri - MaxCompute

Kullanım notları

Senkron çalıştırma için execute_sql('statement'), asenkron çalıştırma için ise run_sql('statement') kullanın. Her iki yöntem de çalışan bir görev örneği döndürür.
MaxCompute, örnek sonuçlarının Arrow biçiminde okunmasını desteklemez.
MaxCompute konsolunda çalışan bazı SQL ifadeleri PyODPS'te çalıştırılamaz. DDL (Veri Tanımlama Dili) ve DML (Veri İşleme Dili) dışındaki ifadeler için uygun yöntemi kullanın:
- run_security_query — GRANT ve REVOKE ifadelerini çalıştırır
- run_xflow veya execute_xflow — Platform for AI (PAI) ifadelerini çalıştırır
SQL işleri, gönderilen iş sayısına göre faturalandırılır. Ayrıntılar için Faturalandırma genel bakış konusuna bakın.

SQL ifadelerini çalıştırma

Hem execute_sql() hem de run_sql() SQL'i MaxCompute'e gönderir. Aradaki fark, kodunuzun sonucu bekleyip beklemediğidir.

import os
from odps import ODPS

# Store credentials in environment variables rather than hardcoding them
o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='<your-endpoint>',
)

# Synchronous: blocks until the SQL job completes
o.execute_sql('select * from table_name')

# Asynchronous: returns immediately; the job runs in the background
instance = o.run_sql('select * from table_name')
print(instance.get_logview_address())  # Get the Logview URL for monitoring
instance.wait_for_success()            # Block until the job finishes

Senkron yöntem, iş tamamlanana kadar geçerli iş parçacığını engeller. Asenkron yöntem ise hemen bir görev örneği döndürerek birden fazla işi paralel olarak göndermenize ve bunları seçici olarak beklemenize olanak tanır.

Asenkron sorguları paralel olarak çalıştırırken, bunları göndermeden önce sorgular arasındaki bağımlılık sırasını anladığınızdan emin olun. Uzun süren asenkron görevlerin mevcut oturumdan daha uzun süre yaşaması gerekiyorsa, bu görevleri ayrı bağlantılara izole edin.

Çalışma zamanı parametrelerini ayarlama

Tek bir çalıştırma için oturum ayarlarını geçersiz kılmak üzere hints parametresini bir dict olarak iletin:

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

Aynı ayarları oturumdaki her çalıştırmaya uygulamak için options.sql.settings değerini genel olarak yapılandırın:

from odps import options

options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('select * from pyodps_iris')  # Picks up the global hints automatically

Sorgu sonuçlarını okuma

Bir sorgu tamamlandıktan sonra sonuçlara erişmek için open_reader kullanın. Yaklaşımı, sorgunun ne döndürdüğüne ve ne kadar veriye ihtiyacınız olduğuna bağlı olarak seçin.

Senaryo	Yaklaşım
Yapılandırılmış veriler (küçük ila orta boyutlu sonuç kümesi)	`open_reader()` ile kayıtlar üzerinde yineleme yapın
Ham metin çıktısı (örneğin, `DESC` komutu)	`reader.raw` değerini okuyun
Tüm sonuçları belleğe yükleme	`reader.to_pandas()`
Paralel indirme ile büyük sonuç kümesi	`reader.to_pandas(n_process=N)`

Yapılandırılmış veriler — kayıtlar üzerinde yineleme yapın:

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

Ham çıktı (örneğin, bir DESC komutundan) — reader.raw değerini okuyun:

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

Instance Tunnel ile Result arayüzü arasında seçim yapma

open_reader iki arka uç desteği sunar:

Arka Uç	Ne Zaman Kullanılır	Nasıl Seçilir
Instance Tunnel	İsteğe bağlı; büyük sonuç kümelerini destekler	`options.tunnel.use_instance_tunnel = True` ayarlayın, ardından `tunnel=True` kullanın
Result arayüzü	Varsayılan veya eski MaxCompute sürümleri ve Instance Tunnel hataları için geri dönüş	`tunnel=False` (veya `options.tunnel.use_instance_tunnel = False` ayarlayın)

PyODPS, Instance Tunnel başarısız olursa otomatik olarak Result arayüzüne düşer ve nedeni açıklayan bir uyarı günlüğe kaydeder. Arka uçları manuel olarak değiştirmek için:

# Use Instance Tunnel explicitly
with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
    for record in reader:
        print(record)

# Use the Result interface explicitly
with o.execute_sql('select * from dual').open_reader(tunnel=False) as reader:
    for record in reader:
        print(record)

İndirilen kayıt sayısını sınırlama

Döndürülen kayıt sayısını sınırlamak için open_reader öğesine bir limit seçeneği iletin veya options.tunnel.limit_instance_tunnel = True ayarlayın. Açık bir yapılandırma olmadan MaxCompute, proje düzeyindeki Tunnel sınırını uygular; bu genellikle indirme başına 10.000 kayıttır.

Sonuçları pandas DataFrame'ine okuma

Tüm sonuçları bir pandas DataFrame'ine yüklemek için okuyucu üzerinde to_pandas() çağrısı yapın:

with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
    pd_df = reader.to_pandas()  # Returns a pandas DataFrame

Çoklu işlemle okumayı hızlandırma

Çoklu işlemle okuma için PyODPS 0.11.3 veya sonraki bir sürüm gerekir.

İndirmeyi birden fazla işleme dağıtmak için to_pandas() öğesine n_process parametresini iletin. SDK, sonuç kümesini partilere böler ve bunları paralel olarak indirir.

import multiprocessing

n_process = multiprocessing.cpu_count()
with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
    pd_df = reader.to_pandas(n_process=n_process)

UDF'ler için kaynak takma adlarını yapılandırma

SQL, bir kaynak dosyasına referans veren kullanıcı tanımlı bir işlev (UDF) çalıştırdığında ve kaynak içeriği değiştiğinde, normalde UDF'yi silip yeniden oluşturmanız gerekir. Bunun yerine, eski kaynak adını yeniyle eşlemek için aliases parametresini kullanın; UDF değişikliklerine gerek kalmaz.

Aşağıdaki örnek, bir kaynak dosyasından bir tamsayı okuyan ve bunu giriş değerine ekleyen bir UDF kurar:

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

UDF'yi değiştirmeden aynı sorguyu farklı bir kaynakla çalıştırmak için aliases parametresini iletin:

res2 = o.create_resource('test_alias_res2', 'file', file_obj='2')

# Map the old resource name to the new one at query time
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])

Etkileşimli ortamlarda SQL çalıştırma

PyODPS, IPython ve Jupyter için parametreli sorguları destekleyen SQL eklentileri içerir. Kurulum talimatları için kullanıcı deneyimi geliştirme belgelerine bakın.

biz_id ayarlama

Bazı durumlarda, SQL ifadelerini gönderirken biz_id değeri iletmeniz gerekebilir. Aksi takdirde, SQL ifadeleri çalıştırıldığında bir hata oluşur. Oturumdaki tüm çalıştırmalara uygulanması için bunu genel olarak ayarlayın:

from odps import options

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