DataWorks:PyODPS 3 düğümü

Giriş

PyODPS, MaxCompute için Python SDK'sıdır. Python kullanarak işler yazmanızı, tabloları ve görünümleri sorgulamanızı ve MaxCompute kaynaklarını yönetmenizi sağlar. Daha fazla bilgi için bkz. PyODPS'e genel bakış. DataWorks'te PyODPS düğümleri, Python görevlerini diğer işlerle birlikte zamanlamanıza ve çalıştırmanıza olanak tanır.

Kullanım notları

• DataWorks kaynak grubundaki bir PyODPS düğümünden üçüncü taraf bir paketi çağırmak için serverless resource group ile birlikte bir custom image kullanın.

  Not

  Kodunuz üçüncü taraf bir pakete başvuran bir kullanıcı tanımlı işlev (UDF) içeriyorsa bu yöntem uygulanmaz. Doğru yordam için bkz. UDF örneği: Python UDF'de üçüncü taraf paket kullanma.

• PyODPS sürümünü yükseltmek için serverless resource group içindeki bir custom image kullanarak /home/tops/bin/pip3 install pyodps==0.12.1 komutunu çalıştırabilirsiniz. 0.12.1 değerini hedef PyODPS sürümüyle değiştirebilirsiniz. Zamanlama için özel kaynak grubu kullanıyorsanız aynı komutu çalıştırmak üzere O&M Assistant kullanın.

• PyODPS işinizin VPC veya IDC ağındaki bir veri kaynağı ya da hizmet gibi belirli bir ağ ortamına erişmesi gerekiyorsa serverless resource group kullanın. Serverless resource group'u hedef ortama bağlama hakkında 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 iki türde sunulur: PyODPS 2 (Python 2) ve PyODPS 3 (Python 3). Python sürümünüzle eşleşen düğüm türünü oluşturun.

• Bir PyODPS düğümünde SQL deyimlerini çalıştırmak Data Map'te doğru veri soyunu oluşturamazsa ilgili DataWorks zamanlama parametrelerini iş kodunda manuel olarak ayarlayın. Veri soyunu görüntüleme hakkında bilgi için bkz. Veri soyunu görüntüleme. Parametreleri ayarlama hakkında bilgi için bkz. Çalışma zamanı parametre ipuçlarını ayarlama. Aşağıdaki örnek kod, çalışma zamanında gerekli parametreleri alır.

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

• Bir PyODPS düğümünün çıkış günlüğü maksimum 4 MB boyutundadır. Günlüğe büyük miktarda veri yazdırmaktan kaçının. Uyarı ve ilerleme bilgilerini çıktı olarak vermeye odaklanın.

Sınırlamalar

• Bir PyODPS düğümünü zamanlama için özel kaynak grubunda çalıştırdığınızda yerel olarak işlenen verilerin 50 MB'ı aşmamasını öneririz. Bu sınır, özel kaynak grubunun teknik özelliklerinden kaynaklanır. Yerel veriler işletim sistemi eşik değerini aşarsa OOM (Got Killed) hatası oluşabilir. PyODPS düğümünde aşırı veri işleme kodu yazmaktan kaçının.

• Bir PyODPS düğümünü serverless resource group üzerinde çalıştırdığınızda veri hacmine göre uygun sayıda CU yapılandırın.

  Not

  Serverless resource group üzerinde tek bir iş en fazla 64 CUs destekler. Ancak iş başlatmayı etkileyebilecek kaynak eksikliklerini önlemek için 16 CUs değerini aşmamanızı öneririz.

• Got killed hatası, işlemin bellek sınırını aştığını gösterir. Yerel veri işlemlerinden kaçının. PyODPS aracılığıyla başlatılan SQL ve DataFrame işleri (to_pandas işlemleri hariç) bu sınırlamaya tabi değildir.

• Kullanıcı tanımlı işlevler içermeyen kodlarda ö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'te varsayılan olarak False olarak ayarlanır. instance tunnel özelliğini genel olarak etkinleştirmek için bu değeri manuel olarak True olarak ayarlayın.

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

  MaxCompute şu anda Python 3.7 kullanmaktadır. Python 3.8'in finally block yapısı gibi diğer Python 3 sürümlerine ait sözdizimini kullanırsanız çalıştırma sırasında hata oluşur. Python 3.7 kullanmanızı öneririz.

• PyODPS 3, serverless resource group üzerinde çalışmayı destekler. Satın alma ve kullanma hakkında bilgi için bkz. Serverless resource group kullanma.

• Tek bir PyODPS düğümü içinde birden fazla Python işini eşzamanlı çalışacak şekilde yapılandıramazsınız.

• Bir PyODPS düğümünde günlük yazdırmak için print kullanın. logger.info kullanımı desteklenmez.

Ön koşullar

MaxCompute hesap motorunu DataWorks çalışma alanınızla ilişkilendirin.

Yordam

1. Kodunuzu PyODPS 3 düğümünün düzenleyici sayfasında geliştirin.

   PyODPS 3 kod örnekleri

   Bir PyODPS düğümü oluşturduktan sonra kodunuzu düzenleyip çalıştırabilirsiniz. PyODPS sözdizimi hakkında daha fazla bilgi için bkz. Temel işlemler. Aşağıdaki örnekler beş yaygın senaryoyu kapsar. İhtiyacınıza uygun olanı seçin.

   ODPS giriş noktası

   DataWorks'teki her PyODPS düğümü, manuel olarak tanımlamanız gerekmeyen global bir ODPS giriş noktası değişkeni olan odps veya o içerir.

   print(odps.exist_table('PyODPS_iris'))

   SQL yürütme

   Bir PyODPS düğümünde SQL deyimlerini çalıştırabilirsiniz. Daha fazla bilgi için bkz. SQL.

   • Varsayılan olarak DataWorks'te instance tunnel devre dışıdır, bu nedenle instance.open_reader Result arabirimini kullanır ve en fazla 10.000 kayıt döndürür. Kayıt sayısını almak için reader.count kullanabilirsiniz. Tüm veriler üzerinde yineleme yapmak için limit seçeneğini devre dışı bırakın. Aşağıdaki ifadeler instance tunnel özelliğini genel olarak etkinleştirir ve limit seçeneğini devre dışı bırakır.

     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 Instance Tunnel.

   • open_reader çağrısına tunnel=True ekleyerek tek bir open_reader çağrısı için de instance tunnel özelliğini etkinleştirebilirsiniz. Çağrı için limit kısıtlamasını devre dışı bırakmak üzere limit=False ekleyebilirsiniz.

     # Use Instance Tunnel for this open_reader call and read all data.
     with instance.open_reader(tunnel=True, limit=False) as reader:

   Çalışma zamanı parametrelerini ayarlama

   • Bir dict olan hints parametresini kullanarak çalışma zamanı parametrelerini ayarlayabilirsiniz. İ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})

   • sql.settings kullanarak global yapılandırmaları ayarlarsanız ilgili ç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')  # Hints are added based on the global settings.

   Yürütme sonuçlarını okuma

   Bir SQL yürütme örneği üzerinde doğrudan open_reader çağrısı yapabilirsiniz. Bu, iki senaryoyu destekler:

   • SQL 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 SQL deyimlerini yürütürseniz SQL yürütmesinin ham sonuçlarını almak için reader.raw özelliğini kullanabilirsiniz.

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

     Not

     Özel zamanlama parametreleri kullanıyorsanız ve bir PyODPS 3 düğümünü doğrudan UI üzerinden çalıştırıyorsanız düğüm çalışma zamanında değişkenleri değiştiremediğinden zaman değerini sabit kodlamalısınız.

   DataFrame

   Verileri DataFrame (Kullanımdan Kaldırıldı) kullanarak da işleyebilirsiniz.

   • Yürütme

     DataWorks ortamında DataFrame işlemleri bir anında yürütme yöntemi çağrısı gerektirir.

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

     Yazdırma sırasında anında yürütmeyi tetiklemeniz gerekiyorsa options.interactive seçeneğini etkinleştirmelisiniz.

     from odps import options
     from odps.df import DataFrame
     options.interactive = True  # Enable the option at the beginning.
     iris = DataFrame(o.get_table('pyodps_iris'))
     print(iris.sepal_width.sum())  # This triggers immediate execution.

   • Ayrıntılı bilgileri yazdırma

     DataWorks'te options.verbose varsayılan olarak etkindir ve bu durum yürütme sırasında Logview URL'si gibi ayrıntılı bilgilerin yazdırılmasına neden olur.

   PyODPS 3 kod geliştirme

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

   1. pyodps_iris örnek tablosunu oluşturarak bir veri kümesi hazırlayın. Daha fazla bilgi için bkz. DataFrame verilerini işleme.

   2. Bir DataFrame oluşturun. Daha fazla bilgi için bkz. MaxCompute tablosundan DataFrame oluşturma.

   3. Aşağıdaki kodu PyODPS düğümüne 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))

   PyODPS işini çalıştırma

   1. Run Configuration bölmesinde Compute Resource bölümünün altında Compute Resource, computing quota ve DataWorks Resource Group öğelerini yapılandırın.

      Not

      • Genel ağ veya VPC üzerinden bir veri kaynağına erişmek için veri kaynağıyla bağlantı testini geçmiş bir zamanlama kaynak grubu kullanmalısınız. Daha fazla bilgi için bkz. Ağ bağlantısı çözümleri.

      • İş gereksinimlerine göre Image bilgilerini yapılandırabilirsiniz.

   2. Araç çubuğundaki parametreler iletişim kutusunda oluşturulan MaxCompute veri kaynağını seçin ve PyODPS işini çalıştırmak için Run öğesine tıklayın.

2. Düğümü periyodik olarak çalıştırmak için iş gereksinimlerinize göre zamanlama özelliklerini yapılandırın. Daha fazla bilgi için bkz. Düğüm zamanlamasını yapılandırma.

   DataWorks'teki SQL düğümlerinden farklı olarak PyODPS düğümleri kodda ${param_name} gibi dizeleri değiştirmez. Bunun yerine kod yürütülmeden önce global değişkenlere args adlı bir sözlük eklenir ve zamanlama parametrelerini buradan alabilirsiniz. Örneğin Parameter bölümünde ds=${yyyymmdd} ayarladıysanız parametre bilgilerini kodda aşağıdaki gibi alabilirsiniz.

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

   Not

   ds adlı bölümü almanız gerekiyorsa aşağıdaki yöntemi kullanabilirsiniz.

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

3. Düğüm yapılandırıldıktan sonra dağıtmanız gerekir. Daha fazla bilgi için bkz. Düğüm ve iş akışı dağıtımı.

4. İş dağıtıldıktan sonra durumunu Operation Center'da görüntüleyebilirsiniz. Daha fazla bilgi için bkz. Operation Center'a giriş.

İlişkili rol kullanarak düğüm çalıştırma

Bir düğümü çalıştırmak için RAM rolü ilişkilendirebilirsiniz. Bu, ince taneli izin kontrolü ve güvenlik yönetimi için düğüm görevlerini belirli bir RAM rolüyle çalıştırmanıza olanak tanır.

Sonraki adımlar

PyODPS Hakkında SSS: PyODPS yürütme sırasında karşılaşılan yaygın sorunlar ve bunların çözülmesi.