MaxCompute:PyODPS 3 işi geliştirme

Ön koşullar

Bir PyODPS 3 düğümü oluşturulmuş olmalıdır. Daha fazla bilgi için bkz. MaxCompute düğümlerini oluşturma ve yönetme.

Arka plan bilgileri

PyODPS, MaxCompute için Python SDK'sıdır. MaxCompute işleri yazmak, tabloları ve görünümleri sorgulamak ve kaynakları yönetmek için bir Python programlama arabirimi sunar. Daha fazla bilgi için bkz. PyODPS. DataWorks içinde, Python işlerini zamanlamak ve çalıştırmak için bir PyODPS düğümü kullanabilir ve bunları diğer iş türleriyle entegre edebilirsiniz.

Kullanım notları

• PyODPS kodunuz üçüncü taraf paketler gerektiriyorsa, bunları bir serverless kaynak grubu ve bir özel görüntü kullanarak yükleyebilirsiniz.

  Not

  Kodunuz üçüncü taraf bir pakete referans veren bir Kullanıcı Tanımlı İşlev (UDF) içeriyorsa bu yöntem desteklenmez. Doğru yapılandırma için bkz. UDF örneği: Python UDF'lerinde üçüncü taraf paketleri kullanma.

• PyODPS sürümünü yükseltmek için serverless bir kaynak grubunda /home/tops/bin/pip3 install pyodps==0.12.1 komutunu çalıştıran bir özel görüntü kullanın (0.12.1 değerini hedef PyODPS sürümüyle değiştirebilirsiniz) veya zamanlama için ayrılmış bir kaynak grubunda aynı komutu çalıştırmak üzere O&M Assistant kullanın.

• PyODPS işinizin bir VPC veya şirket içi veri merkezi (IDC) gibi özel bir ağ ortamına erişmesi gerekiyorsa serverless bir kaynak grubu kullanın ve kaynak grubu ile hedef ortam arasında bir ağ bağlantısı kurun. Daha fazla bilgi için bkz. Ağ bağlantısı çözümleri.

• PyODPS sözdizimi hakkında daha fazla bilgi için bkz. PyODPS belgeleri.

• PyODPS düğümleri PyODPS 2 ve PyODPS 3 olmak üzere iki türde bulunur. Farklı temel Python sürümleri kullanırlar: PyODPS 2 düğümleri Python 2, PyODPS 3 düğümleri ise Python 3 kullanır. Python sürümünüzle eşleşen düğüm türünü oluşturduğunuzdan emin olun.

• Bir PyODPS düğümünde SQL yürütülmesi doğru veri soy ağacını oluşturamaz ve bu soy ağacı Data Map içinde görünmezse sorunu kodunuzda DataWorks zamanlama ve çalışma zamanı parametrelerini manuel olarak ayarlayarak çözün. Veri soy ağacını nasıl görüntüleyeceğinizi öğrenmek için bkz. Veri soy ağacını görüntüleme. Parametre ayarları için bkz. Çalışma zamanı parametresi ipuçlarını ayarlama. Gerekli çalışma zamanı parametrelerini aşağıdaki örnek kodu kullanarak elde edebilirsiniz:

  import os
  ...
  # get DataWorks sheduler runtime parameters
  skynet_hints = {}
  for k, v in os.environ.items():
      if k.startswith('SKYNET_'):
          skynet_hints[k] = v
  ...
  # setting hints while submiting a task
  o.execute_sql('INSERT OVERWRITE TABLE XXXX SELECT * FROM YYYY WHERE ***', hints=skynet_hints)
  ...

• Bir PyODPS düğümünün çıktı günlük boyutu maksimum 4 MB'dır. Büyük veri sonuçlarını günlüğe yazdırmaktan kaçının. Bunun yerine yalnızca temel uyarı ve ilerleme bilgilerini çıktılar.

Sınırlamalar

• Zamanlama için ayrılmış bir kaynak grubunda bir PyODPS düğümü çalıştırırken 50 MB'tan fazla yerel veriyi işlemeyin. Bu sınırlama, zamanlama için ayrılmış kaynak grubunun kaynak teknik özelliklerinden kaynaklanır. İşletim sisteminin eşiğini aşan büyük miktarda yerel verinin işlenmesi, bellek yetersizliği (OOM) hatasına neden olabilir; bu durum Got Killed mesajıyla belirtilir. PyODPS düğümüne doğrudan kapsamlı veri işleme kodu yazmaktan kaçının.

• Serverless bir kaynak grubu kullanarak bir PYODPS düğümü çalıştırdığınızda, düğümün işlemesi gereken veri miktarına bağlı olarak CU'ları yapılandırabilirsiniz.

  Not

  Serverless bir kaynak grubunda bir görev çalıştırdığınızda tek bir görev maksimum 64CU yapılandırmasını destekler ancak aşırı CU değerinden kaynaklanan kaynak eksikliklerini önlemek ve görev başlatmayı etkilememek için 16CU değerini aşmamanız önerilir.

• Got killed hatası, memory kullanımının sınırı aştığını ve işlemin sonlandırıldığını gösterir. Bunu önlemek için yerel veri işlemlerinden kaçının. Bu sınırlama, PyODPS aracılığıyla başlatılan SQL veya DataFrame işleri (to_pandas hariç) için geçerli değildir.

• Özel işlevler içermeyen kodlar için önceden yüklenmiş Numpy ve Pandas kitaplıklarını kullanabilirsiniz. İkili kod içeren diğer üçüncü taraf paketler desteklenmez.

• Uyumluluk nedenleriyle, options.tunnel.use_instance_tunnel DataWorks içinde varsayılan olarak False olarak ayarlanır. instance tunnel özelliğini genel olarak etkinleştirmeniz gerekirse bu değeri manuel olarak True olarak ayarlamanız gerekir.

• Bytecode tanımı, Python 3.8 ve Python 3.7 gibi Python 3'ün alt sürümleri arasında farklılık gösterir.

  MaxCompute şu anda Python 3.7 kullanmaktadır. Python 3.8'deki finally bloğu gibi diğer Python 3 sürümlerinin sözdizimini kullanırsanız yürütme hatası oluşur. Python 3.7 kullanmanız önerilir.

• PyODPS 3, bir serverless kaynak grubunda çalışmayı destekler. Bir tane satın almak ve kullanmak için bkz. Serverless kaynak gruplarını kullanma.

• Tek bir PyODPS düğümü içinde birden çok Python işini eş zamanlı olarak çalıştırmak desteklenmez.

Kodu düzenleme: Temel örnek

Bir PyODPS düğümü oluşturduktan sonra kodunuzu düzenleyebilir ve çalıştırabilirsiniz. PyODPS sözdizimi hakkında daha fazla bilgi için bkz. Temel işlemlere genel bakış.

• ODPS giriş noktası

  DataWorks PyODPS düğümü, ODPS giriş noktası olarak odps veya o adlı bir global değişken sağlar. Bunu manuel olarak tanımlamanız gerekmez.

  print(odps.exist_table('PyODPS_iris'))

• SQL yürütme

  Bir PyODPS düğümünde SQL deyimleri yürütebilirsiniz. Daha fazla bilgi için bkz. SQL.

  • Varsayılan olarak, DataWorks içinde instance tunnel devre dışı bırakılır. Bu, instance.open_reader öğesinin en fazla 10.000 kayıt okuyan Result arabirimini kullandığı anlamına gelir. Kayıt sayısını almak için reader.count kullanabilirsiniz. Tüm veriler üzerinde yineleme yapmak için limit özelliğini devre dışı bırakmalısınız. Instance tunnel özelliğini genel olarak etkinleştirmek ve limit özelliğini devre dışı bırakmak için aşağıdaki deyimleri kullanın.

    options.tunnel.use_instance_tunnel = True
    options.tunnel.limit_instance_tunnel = False  # Disable the limit to read all data.
    with instance.open_reader() as reader:
      # All data can be read through the instance tunnel.

  • Geçerli open_reader çağrısı için instance tunnel özelliğini etkinleştirmek üzere open_reader öğesine tunnel=True ekleyebilirsiniz. Benzer şekilde, geçerli çağrı için limit kısıtlamasını devre dışı bırakmak için limit=False ekleyebilirsiniz.

    # Use the Instance Tunnel interface for the current open_reader operation to read all data.
    with instance.open_reader(tunnel=True, limit=False) as reader:

• Çalışma zamanı parametreleri

  • Bir dict olan hints parametresini kullanarak çalışma zamanı parametrelerini ayarlayın. İpuçları hakkında daha fazla bilgi için bkz. SET işlemleri.

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

  • Global yapılandırmada sql.settings ayarlarsanız bu çalışma zamanı parametreleri her yürütmeye eklenir.

    from odps import options
    options.sql.settings = {'odps.sql.mapper.split.size': 16}
    o.execute_sql('select * from PyODPS_iris')  # This call includes hints from the global configuration.

• Yürütme sonuçları

  Bir SQL yürütme örneği, aşağıdaki iki senaryoda doğrudan open_reader işlemini yürütebilir:

  • SQL deyimi yapılandırılmış veri döndürür.

    with o.execute_sql('select * from dual').open_reader() as reader:
    	for record in reader:  # Process each record.

  • desc gibi deyimleri yürütürken ham SQL yürütme sonucunu reader.raw özelliğini kullanarak alabilirsiniz.

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

    Not

    Özel zamanlama parametreleri kullanıyorsanız bir PyODPS 3 düğümünü sayfadan doğrudan çalıştırdığınızda saati sabit kodlamanız gerekir. PyODPS düğümü bu değeri doğrudan değiştiremez.

• DataFrame

  Verileri bir DataFrame (önerilmez) kullanarak da işleyebilirsiniz.

  • Yürütme

    DataWorks ortamında, DataFrame işlemlerinin açıkça bir hemen yürütülen yöntem çağrılarak tetiklenmesi gerekir.

    from odps.df import DataFrame
    iris = DataFrame(o.get_table('pyodps_iris'))
    for record in iris[iris.sepal_width < 3].execute():  # Call an immediately executed method to process each record.

    Yazdırma sırasında hemen yürütmeyi tetiklemeniz gerekirse options.interactive özelliğini etkinleştirmelisiniz.

    from odps import options
    from odps.df import DataFrame
    options.interactive = True  # Enable the switch at the beginning.
    iris = DataFrame(o.get_table('pyodps_iris'))
    print(iris.sepal_width.sum())  # An immediate execution is triggered when printing.

  • Ayrıntılı bilgileri yazdırma

    options.verbose seçeneğini ayarlayın. Bu seçenek DataWorks içinde varsayılan olarak etkindir ve yürütme sırasında Logview URL'si gibi ayrıntılı bilgileri yazdırır.

Örnek

Aşağıdaki örnek, bir PyODPS düğümünün nasıl kullanılacağını göstermektedir:

1. Veri kümesini hazırlayın ve pyodps_iris örnek tablosunu oluşturun. Ayrıntılar için bkz. DataFrame kullanarak verileri işleme.

2. Bir DataFrame oluşturun. Ayrıntılar için bkz. MaxCompute tablosundan DataFrame oluşturma.

3. PyODPS düğümüne aşağıdaki kodu girin ve çalıştırın.

   from odps.df import DataFrame
   # Create a DataFrame from an ODPS table.
   iris = DataFrame(o.get_table('pyodps_iris'))
   print(iris.sepallength.head(5))

   Aşağıdaki sonuç döndürülür:

      sepallength
   0          4.5
   1          5.5
   2          4.9
   3          5.0
   4          6.0

Kodu düzenleme: Gelişmiş örnek

Düğümün periyodik olarak çalışması gerekiyorsa zamanlama özelliklerini tanımlamanız gerekir. Daha fazla bilgi için bkz. Bir düğüm için zamanlama özelliklerini yapılandırma.

print('ds=' + args['ds'])
ds=20161116

o.get_table('table_name').get_partition('ds=' + args['ds'])

Diğer kullanım durumları için PyODPS işleri geliştirmek hakkında daha fazla bilgi için aşağıdaki konulara bakın:

• PyODPS'te üçüncü taraf paketleri kullanma.

• PyODPS düğümünde Jieba ile Çince metinleri bölümlere ayırma.

• Verileri yerel bilgisayara indirmekten kaçınmak için PyODPS düğümü kullanma.

Sonraki adımlar

• Özel bir Shell betiğinin başarıyla çalışıp çalışmadığını belirleme: Özel bir Python betiğinin başarıyla çalışıp çalışmadığını belirleme mantığı, bir Shell betiğiyle aynıdır. Doğrulama için bu yöntemi kullanabilirsiniz.

• İş dağıtma: Standart modda bir çalışma alanı kullanıyorsanız işin periyodik olarak çalışabilmesi için onu üretim ortamına dağıtmanız gerekir.

• Periyodik olarak çalışan işler için O&M: Bir iş üretim ortamına dağıtıldıktan ve zamanlandıktan sonra Operation Center içinde iş üzerinde O&M gerçekleştirebilirsiniz.

• PyODPS SSS: Sorunları hızlı bir şekilde gidermenize yardımcı olmak için PyODPS işlerini çalıştırma hakkındaki yaygın soruların yanıtlarını bulun.

SSS

S: Lark gibi üçüncü taraf bir API'den veri toplamak ve bunu DataWorks'e aktarmak için bir PyODPS3 düğümü kullanıyorum. Kod, yerel geliştirme ortamımda sorunsuz çalışıyor ancak üretim ortamına gönderildiğinde ve Operation Center içinde yürütüldüğünde yanıt zaman aşımı hatası bildiriyor. Neden?

C: Management Center > Security Settings > Workspace. Sandbox izin listesinde, PyODPS 3 işine erişim izni vermek için üçüncü taraf API'nin alan adını ekleyin. Örnek:

Lark API alan adını open.feishu.cn ekleyin ve bağlantı noktasını 443 olarak ayarlayın.