DataWorks:PyODPS 2 düğümü

Genel bakış

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ğlayan sade bir programlama arayüzü sunar. Daha fazla bilgi için bkz. PyODPS genel bakışı. DataWorks'te Python görevlerini zamanlamak, çalıştırmak ve bunları diğer işlerle entegre etmek için bir PyODPS düğümü kullanabilirsiniz.

Notlar

• Bir DataWorks kaynak grubunda PyODPS düğümü çalıştırırken kodunuz üçüncü taraf paketler gerektiriyorsa bu paketleri yüklemek için bir serverless kaynak grubu ve bir özel görüntü kullanın.

  Not

  Kodunuz üçüncü taraf bir pakete başvuran bir UDF kullanıyorsa bu yöntem desteklenmez. Bu senaryonun nasıl ele alınacağı hakkında bilgi için bkz. UDF örneği: Python UDF'lerinde üçüncü taraf paketleri kullanma.

• PyODPS görevinizin bir VPC'deki veya şirket içi veri merkezindeki bir veri kaynağı ya da hizmet gibi belirli bir ağ ortamına erişmesi gerekiyorsa serverless kaynak grubu kullanın ve gerekli ağ bağlantısını kurmak için Ağ bağlantısı çözümleri bölümüne bakın.

• PyODPS sözdizimi hakkında daha fazla bilgi için PyODPS belgelerine bakın.

• PyODPS düğümleri PyODPS 2 ve PyODPS 3 olmak üzere iki türde sunulur. Temel Python sürümlerine göre farklılık gösterirler. PyODPS 2 düğümleri Python 2, PyODPS 3 düğümleri ise Python 3 kullanır. Kullandığınız Python sürümüne karşılık gelen düğüm türünü oluşturun.

• Bir PyODPS düğümünde SQL yürütmenin veri soyunu oluşturamaması durumunda Veri Haritası'nda veri soyu görüntülenemez. Sorunu çözmek için görev kodunda DataWorks zamanlama parametrelerini manuel olarak ayarlayabilirsiniz. Veri soyunu görüntülemek için bkz. Veri soyunu görüntüleme. Parametre ayarları hakkında bilgi için bkz. Çalışma zamanı parametre 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ü için çıkış günlüğünün maksimum boyutu 4 MB'dır. Günlüğe doğrudan büyük miktarda veri yazdırmaktan kaçının. Bunun yerine yalnızca temel uyarıları ve ilerleme güncellemelerini günlüğe kaydedin.

Sınırlamalar

• Bir PyODPS düğümünü özel bir zamanlama kaynak grubunda çalıştırdığınızda düğüm içinde yerel olarak işlenen verilerin 50 MB'ı aşmamasını öneririz. Bu işlem, özel zamanlama kaynak grubunun teknik özellikleriyle sınırlıdır. İşletim sistemi eşiğini aşan çok fazla yerel verinin işlenmesi OOM (işlem sonlandırıldı) hatasına neden olabilir. PyODPS düğümlerinde çok fazla veri işleme kodu yazmaktan kaçının.

• Bir PyODPS düğümünü serverless bir kaynak grubunda çalıştırdığınızda işlenecek veri hacmine bağlı olarak PyODPS düğümü için CU tahsisini yapılandırabilirsiniz.

  Not

  Bu görevi serverless bir kaynak grubunda çalıştırdığınızda tek bir görev için maksimum yapılandırma 64CU değerindedir. Ancak aşırı CU tahsisinden kaynaklanan yetersiz kaynak sorunlarını ve bunun görev başlatmayı etkilemesini önlemek amacıyla 16CU değerini aşmamanız önerilir.

• Got Killed hatasıyla karşılaşırsanız bu durum bellek kullanımının sınırı aştığını ve işlemin sonlandırıldığını gösterir. Bu nedenle yerel veri işlemlerinden mümkün olduğunca kaçının. PyODPS aracılığıyla başlatılan SQL ve DataFrame görevleri (to_pandas hariç) bu sınırlamaya tabi değildir.

• UDF olmayan kodlar önceden yüklü NumPy ve Pandas paketlerini kullanabilir. İkili kod içeren diğer üçüncü taraf paketler desteklenmez.

• Uyumluluk nedenleriyle DataWorks'te options.tunnel.use_instance_tunnel varsayılan olarak False olarak ayarlanır. Instance Tunnel'ı genel olarak etkinleştirmeniz gerekirse bu değeri manuel olarak True olarak ayarlamanız gerekir.

• PyODPS 2 düğümleri için temel Python sürümü 2,7'dir.

• Bir PyODPS düğümü içinde birden çok Python görevinin eş zamanlı olarak çalıştırılması desteklenmez.

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

Önkoşullar

MaxCompute hesaplama motorunu DataWorks çalışma alanınızla ilişkilendirme.

Yordam

1. PyODPS 2 düğümü düzenleme sayfasında aşağıdaki geliştirme işlemlerini gerçekleştirin.

   PyODPS 2 kod örnekleri

   Bir PyODPS düğümü oluşturduktan sonra kodu düzenleyebilir ve çalıştırabilirsiniz. PyODPS sözdizimi hakkında daha fazla bilgi için bkz. PyODPS belgeleri. Bu konuda aşağıdaki beş kod örneği açıklanmaktadır. İş ihtiyaçlarınıza göre örnekleri seçebilirsiniz.

   ODPS girişi

   DataWorks PyODPS düğümlerinde ODPS girişi olarak global bir değişken odps veya o bulunur. ODPS girişini manuel olarak tanımlamanız gerekmez.

   print(odps.exist_table('PyODPS_iris'))

   SQL yürütme

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

   • Varsayılan olarak DataWorks'te instance tunnel etkin değildir. Bu durum instance.open_reader işleminin varsayılan olarak Result arabirimini (en fazla 10.000 kayıt) kullandığı anlamına gelir. Kayıt sayısını almak için reader.count kullanabilirsiniz. Tüm veriler üzerinde yineleme yapmanız gerekiyorsa limit kısıtlamasını devre dışı bırakmalısınız. Instance Tunnel işlevini genel olarak etkinleştirmek ve limit kısıtlamasını devre dışı bırakmak için aşağıdaki deyimleri kullanabilirsiniz.

     options.tunnel.use_instance_tunnel = True
     options.tunnel.limit_instance_tunnel = False  # Disable limit restriction, read all data.
     
     with instance.open_reader() as reader:
       # Use Instance Tunnel to read all data.

   • Ayrıca tunnel=True parametresini open_reader çağrısına ekleyerek instance tunnel işlevini yalnızca mevcut open_reader çağrısı için etkinleştirebilirsiniz. Yine aynı şekilde limit=False parametresini ekleyerek limit kısıtlamasını yalnızca mevcut çağrı için devre dışı bırakabilirsiniz.

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

   Çalışma zamanı parametrelerini ayarlama

   • dict türünde olan hints parametresini yapılandırarak çalışma zamanı parametrelerini ayarlayabilirsiniz. Hints parametresi 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 ayarını genel olarak yapılandırdıktan sonra her görev çalıştırıldığında ilgili çalışma zamanı parametrelerini eklemeniz gerekir.

     from odps import options
     options.sql.settings = {'odps.sql.mapper.split.size': 16}
     o.execute_sql('select * from PyODPS_iris')  # Add hints based on global configurationhints。

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

   SQL çalıştıran bir örnek doğrudan open_reader işlemini gerçekleştirebilir. Aşağıdaki iki senaryo mümkündür:

   • 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 eachrecord。

   • SQL deyimi desc gibi bir deyim olabilir. Ham SQL yürütme sonucunu 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 bir PyODPS düğümü parametreleri doğrudan değiştiremediğinden sayfada bir PyODPS 2 düğümünü doğrudan tetiklediğinizde saat değerlerini sabit kodlamanız gerekir.

   DataFrame

   Verileri işlemek için DataFrame de kullanabilirsiniz.

   • Yürütme

     DataWorks ortamında DataFrame yürütmesi anında yürütmeyi tetikleyen yöntemlerin açıkça çağrılmasını gerektirir.

     from odps.df import DataFrame
     iris = DataFrame(o.get_table('pyodps_iris'))
     for record in iris[iris.sepal_width < 3].execute():  # Execute immediately and process each Record.

     Print kullanırken anında yürütmeyi tetiklemek istiyorsanız options.interactive seçeneğini etkinleştirmeniz gerekir.

     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())  # Executes immediately on print.

   • Ayrıntılı bilgileri yazdırma

     options.verbose seçeneğini yapılandırabilirsiniz. DataWorks'te bu seçenek varsayılan olarak etkindir ve yürütme sırasında Logview URL'si gibi ayrıntılı bilgiler yazdırılır.

   PyODPS 2 kod geliştirme

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

   1. Bir veri kümesi hazırlayın ve pyodps_iris örnek tablosunu oluşturun. Daha fazla bilgi için bkz. DataFrame oluşturma.

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

   3. PyODPS düğümüne aşağıdaki kodu girin.

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

   Bir PyODPS görevini çalıştırma

   1. Run Configuration bölümünde Compute Resource, Compute Resource and Quota ve DataWorks Resource Group ayarlarını yapılandırın.

      Not

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

      • Görev gereksinimlerine göre Image bilgilerini yapılandırabilirsiniz.

   2. Araç çubuğunda PyODPS görevini çalıştırmak için Run öğesini tıklayın.

2. Düğüm görevini periyodik olarak çalıştırmanız gerekiyorsa iş ihtiyaçlarınıza göre zamanlama ayarlarını yapılandırın. Daha fazla bilgi için bkz. Zamanlama ayarlarını yapılandırma.

   DataWorks'teki SQL düğümlerinin aksine PyODPS düğümleri kodu etkilememek için kod içindeki ${param_name} gibi dizeleri değiştirmez. Bunun yerine kodu yürütmeden önce global değişkenlere args adlı bir dict eklenir ve zamanlama parametreleri buradan alınabilir. Örneğin Parameter bölümünde ds=${yyyymmdd} ayarını yaptıysanız parametreyi kodunuzda 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 görevi yapılandırmasını tamamladıktan sonra düğümü dağıtmanız gerekir. Daha fazla bilgi için bkz. Düğüm dağıtma.

4. Görevi dağıttıktan sonra Operation Center'da zamanlanmış görevin çalışma durumunu görüntüleyebilirsiniz. Daha fazla bilgi için bkz. Zamanlanmış görevleri görüntüleme.

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

Bir düğümü çalıştırmak için bir RAM rolü ilişkilendirebilirsiniz. Bu sayede ayrıntılı izin denetimi ve güvenlik yönetimi için belirli bir RAM rolüyle düğüm görevleri çalıştırabilirsiniz.

Sonraki adım

PyODPS SSS: PyODPS yürütme sırasındaki yaygın sorunları öğrenerek hızlı sorun giderme işlemleri gerçekleştirebilirsiniz.