MaxCompute:Python 3 kullanıcı tanımlı tablo değerli işlev (UDTF)

Python 3'ü etkinleştirin

Varsayılan olarak MaxCompute projeleri UDF'ler için Python 2 kullanır. Python 3 kullanmak için SQL ifadenizden önce aşağıdaki oturum düzeyindeki komutu ekleyin ve ikisini birlikte gönderin:

set odps.sql.python.version=cp37;

UDTF kod yapısı

Python 3'te UDTF kodu yazmak için MaxCompute Studio aracını kullanın. Bir UDTF dört bileşenden oluşur:

# Import the function signature module and the base class.
from odps.udf import annotate
from odps.udf import BaseUDTF

# Function signature: takes a STRING, returns a STRING.
@annotate('string -> string')

# Custom Python class derived from BaseUDTF.
class Explode(BaseUDTF):

    def process(self, arg):
        props = arg.split(',')
        for p in props:
            self.forward(p)

Sınırlamalar

Python 3, Python 2 ile uyumlu değildir. Tek bir SQL ifadesinde Python 2 UDTF'leri ile Python 3 UDTF'leri karıştırılamaz.

Python 2 UDTF'lerini geçirin

Python 2'nin kullanım ömrü sona ermiştir. Mevcut Python 2 UDTF'lerinizi proje durumunuza göre geçirin:

• Yeni proje veya ilk Python UDTF: Tüm Python UDTF'lerini en baştan Python 3 ile yazın.

• Çok sayıda Python 2 UDTF'i içeren mevcut proje: Kesintiyi önlemek için kademeli olarak geçiş yapın. Aşağıdaki yaklaşımlardan birini seçin:

  • Yeni UDTF'leri Python 3 ile yazın ve bu yeni UDTF'leri kullanan işler için Python 3'ü oturum düzeyinde etkinleştirin. Ayrıntılar için bkz. Python 3'ü etkinleştirin.

  • Mevcut Python 2 UDTF'lerini hem Python 2 hem de Python 3 ile uyumlu olacak şekilde yeniden yazın. Rehberlik için Python 2 Kodunu Python 3'e Taşıma konusuna bakın.

Bir UDTF birden fazla MaxCompute projesinde paylaşılıyorsa, Python 2 kullanan projelerin bozulmasını önlemek amacıyla işlevi hem Python 2 hem de Python 3 ile uyumlu hale getirin.

Üçüncü taraf kitaplıklar

NumPy, MaxCompute Python 3 çalışma zamanı ortamında bulunmaz. Bir UDTF'de NumPy kullanmak için bir NumPy wheel paketini kaynak olarak manuel şekilde yükleyin. Python Package Index (PyPI) veya bir görüntüden beklenen dosya adı şöyledir:

numpy-<Version>-cp37-cp37m-manylinux1_x86_64.whl

Paket yükleme talimatları için bkz. Kaynak işlemleri veya Python UDF'lerinde üçüncü taraf paketlere referans verme.

İşlev imzaları ve veri türleri

İşlev imzası, bir UDTF'nin giriş parametrelerinin ve dönüş değerlerinin veri türlerini bildirir. MaxCompute, semantik ayrıştırma sırasında imzayı doğrular ve gerçek türler eşleşmiyorsa hata döndürür.

@annotate('arg_type_list -> type_list')

Kaynaklara referans verme

Bir Python UDTF'de dosya ve tablolara odps.distcache modülünü kullanarak referans verin.

odps.distcache.get_cache_file(resource_name)

Bir dosya kaynağının içeriğini döndürür.

• resource_name: MaxCompute projenizdeki mevcut bir dosya kaynağının adı. Ad geçersizse veya dosya yoksa hata döndürür.

• Dosya benzeri bir nesne döndürür. İşiniz bittiğinde dosya tanıtıcısını serbest bırakmak için nesne üzerinde close() çağrısı yapın.

• UDTF oluştururken dosya kaynağını bildirin. Bu bildirimi atlarsanız UDTF çağrıldığında hata döner.

odps.distcache.get_cache_table(resource_name)

Bir tablo kaynağının içeriğini döndürür.

• resource_name: MaxCompute projenizdeki mevcut bir tablo kaynağının adı. Ad geçersizse veya tablo yoksa hata döndürür.

• Bir oluşturucu (generator) döndürür. Üzerinde yineleme yapıldığında her satır için bir kayıt elde edilir; her kayıt bir ARRAY'dir.

Aşağıdaki örnek, bir JSON dosyasından ve bir tablo kaynağından veri okur, ardından bir arama işlemine dayalı olarak satırları çıktılar:

from odps.udf import annotate
from odps.udf import BaseUDTF
from odps.distcache import get_cache_file
from odps.distcache import get_cache_table

@annotate('string -> string, bigint')
class UDTFExample(BaseUDTF):

    def __init__(self):
        import json
        # Load the JSON file resource into a dict.
        cache_file = get_cache_file('test_json.txt')
        self.my_dict = json.load(cache_file)
        cache_file.close()

        # Append records from the table resource into the dict.
        records = list(get_cache_table('table_resource1'))
        for record in records:
            self.my_dict[record[0]] = record[1]

    def process(self, pageid):
        # For each input pageid, forward all associated adid values.
        for adid in self.my_dict[pageid]:
            self.forward(pageid, adid)

Python 3 UDTF çağırma

geliştirme sürecini izleyerek bir Python 3 UDTF geliştirdikten sonra bunu MaxCompute SQL'den çağırın.

• Proje içinde: UDTF'yi yerleşik işlev gibi çağırın.

• Projeler arasında: Başka bir projedeki UDTF'ye proje adını önek olarak kullanarak referans verin:

  SELECT B:udf_in_other_project(arg0, arg1) AS res FROM table_t;

  Projeler arası kaynak paylaşımı kurulumu için bkz. Projeler arasında paket tabanlı kaynak paylaşımı.

MaxCompute Studio'da Python 3 UDTF geliştirmek ve test etmek için bkz. Python UDF geliştirme.