Execute and obtain results - MaxCompute - Alibaba Cloud Dokümantasyon Merkezi

Bu konuda, DataFrame işlemleri için kullanabileceğiniz çalıştırma yöntemleri açıklanmaktadır.

Ön koşullar

Aşağıdaki gereksinimlerin karşılandığından emin olun:

pyodps_iris adlı bir örnek tablo hazırlanmıştır. Daha fazla bilgi için bkz. DataFrame veri işleme.
Bir DataFrame nesnesi oluşturulmuştur. Daha fazla bilgi için DataFrame nesnesi oluşturma konusundaki "MaxCompute tablosundan DataFrame nesnesi oluşturma" bölümüne bakın.

Ertelenmiş çalıştırma

DataFrame işlemleri, yalnızca execute yöntemini açıkça çağırdığınızda veya dahili olarak execute yöntemini çağıran yöntemleri kullandığınızda yürütülür. Aşağıdaki tabloda, execute yöntemini dahili olarak çağıran yöntemler listelenmektedir.

Yöntem	Açıklama	Dönüş değeri
persist	Çalıştırma sonuçlarını MaxCompute tablolarına kaydeder.	PyODPS DataFrame
execute	İşlemleri yürütür ve tüm sonuçları döndürür.	ResultFrame
head	İşlemleri yürütür ve sonuç verilerinin ilk N satırını döndürür.	ResultFrame
tail	İşlemleri yürütür ve sonuç verilerinin son N satırını döndürür.	ResultFrame
to_pandas	Bir Collection nesnesini pandas DataFrame nesnesine veya bir Sequence nesnesini Series nesnesine dönüştürür. wrap parametresi True olarak ayarlanırsa bir PyODPS DataFrame nesnesi döndürülür.	wrap parametresi True olarak ayarlanırsa bir PyODPS DataFrame nesnesi döndürülür. wrap parametresi False olarak ayarlanırsa bir pandas DataFrame nesnesi döndürülür. wrap parametresinin varsayılan değeri False'tur.
plot, hist ve boxplot	Çizim yöntemleridir.	Yok

Not

Etkileşimli bir ortamda, PyODPS DataFrame sonuç verilerini görüntülerken veya repr yöntemini çağırırken otomatik olarak execute yöntemini çağırır. execute yöntemini manuel olarak çağırmanız gerekmez.

Örnekler

# In a non-interactive environment, you need to manually call the execute method.
print(iris[iris.sepallength < 5][:5].execute())

# In an interactive environment, the system automatically calls the execute method.
print(iris[iris.sepallength < 5][:5])

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

   sepallength  sepalwidth  petallength  petalwidth         name
0          4.9         3.0          1.4         0.2  Iris-setosa
1          4.7         3.2          1.3         0.2  Iris-setosa
2          4.6         3.1          1.5         0.2  Iris-setosa
3          4.6         3.4          1.4         0.3  Iris-setosa
4          4.4         2.9          1.4         0.2  Iris-setosa

Etkileşimli bir ortamda sistemin execute yöntemini otomatik olarak çağırmasını devre dışı bırakmanız gerekiyorsa manuel işlemler yapmanız gerekir. Aşağıdaki kod bir örneği göstermektedir:

from odps import options
options.interactive = False

print(iris[iris.sepallength < 5][:5])

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

Collection: ref_0
  odps.Table
    name: hudi_mc_0612.`iris3`
    schema:
      sepallength           : double      # Sepal length (cm)
      sepalwidth            : double      # Sepal width (cm)
      petallength           : double      # Petal length (cm)
      petalwidth            : double      # Petal width (cm)
      name                  : string      # Type
Collection: ref_1
  Filter[collection]
    collection: ref_0
    predicate:
      Less[sequence(boolean)]
        sepallength = Column[sequence(float64)] 'sepallength' from collection ref_0
        Scalar[int8]
          5
Slice[collection]
  collection: ref_1
  stop:
    Scalar[int8]
      5

Otomatik çağrıları devre dışı bıraktıktan sonra, repr nesnesi görüntülendiğinde tüm soyut sözdizimi ağacı (AST) gösterilir. Bu durumda, yöntemi kullanmanız gerektiğinde execute yöntemini manuel olarak çağırmanız gerekir.

Çalıştırma sonuçlarını alma

execute veya head yöntemini çağırdıktan sonra bir ResultFrame döndürülürse, ResultFrame'den sonuçları alabilirsiniz.

Not

ResultFrame bir sonuç kümesidir ve sonraki hesaplamalarda kullanılamaz.

ResultFrame'deki tüm kayıtları yinelemeli olarak alabilirsiniz. Aşağıdaki kod bir örneği göstermektedir:

result = iris.head(3)
for r in result:
    print(list(r))

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

[4.9, 3.0, 1.4, 0.2, 'Iris-setosa']
[4.7, 3.2, 1.3, 0.2, 'Iris-setosa']
[4.6, 3.1, 1.5, 0.2, 'Iris-setosa']

pandas yüklüyse, bir ResultFrame pandas DataFrame'e veya pandas arka ucunu kullanan bir PyODPS DataFrame'e dönüştürülebilir.

# Return a pandas DataFrame. 
pd_df = iris.head(3).to_pandas()

# Return a PyODPS DataFrame that uses the pandas backend. 
wrapped_df = iris.head(3).to_pandas(wrap=True)

Sonuçları MaxCompute tablolarına kaydetme

Bir Collection nesnesi için yeni bir DataFrame nesnesi döndürmek üzere persist yöntemini çağırabilirsiniz. persist yöntemi, tablo adını parametre olarak alır.

iris2 = iris[iris.sepalwidth < 2.5].persist('pyodps_iris')
print(iris2.head(5))

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

   sepallength  sepalwidth  petallength  petalwidth             name
0          4.5         2.3          1.3         0.3      Iris-setosa
1          5.5         2.3          4.0         1.3  Iris-versicolor
2          4.9         2.4          3.3         1.0  Iris-versicolor
3          5.0         2.0          3.5         1.0  Iris-versicolor
4          6.0         2.2          4.0         1.0  Iris-versicolor

Bölümlendirilmiş bir tablo oluşturmak için persist yönteminde partitions parametresini belirtebilirsiniz. Tablo, partitions ile belirtilen sütunlara göre bölümlendirilir.

iris3 = iris[iris.sepalwidth < 2.5].persist('pyodps_iris_test', partitions=['name'])
print(iris3.data)

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

odps.Table
  name: odps_test_sqltask_finance.`pyodps_iris`
  schema:
    sepallength           : double
    sepalwidth            : double
    petallength           : double
    petalwidth            : double
  partitions:
    name                  : string

Verileri mevcut bir tablonun bölümüne yazmak için persist yönteminde partition parametresini belirtebilirsiniz. partition parametresi, verilerin yazılacağı bölümü belirtir. Örneğin, partition parametresini ds=****** olarak ayarlayın. Tablo, DataFrame nesnesinin tüm sütunlarını içermeli ve sütunlar aynı türde olmalıdır. drop_partition ve create_partition parametreleri yalnızca partition parametresi belirtildiğinde geçerlidir. drop_partition parametresi, belirtilen bölüm varsa silinip silinmeyeceğini belirtir. create_partition parametresi, belirtilen bölüm yoksa oluşturulup oluşturulmayacağını belirtir.
```
print(iris[iris.sepalwidth < 2.5].persist('pyodps_iris_partition', partition='ds=test', drop_partition=True, create_partition=True).head(5))
```
Aşağıdaki sonuç döndürülür:
```
   sepallength  sepalwidth  petallength  petalwidth             name    ds
0          4.5         2.3          1.3         0.3      Iris-setosa  test
1          5.5         2.3          4.0         1.3  Iris-versicolor  test
2          4.9         2.4          3.3         1.0  Iris-versicolor  test
3          5.0         2.0          3.5         1.0  Iris-versicolor  test
4          6.0         2.2          4.0         1.0  Iris-versicolor  test
```

Verileri bir tabloya yazarken tablonun yaşam süresini (TTL) belirtebilirsiniz. Örneğin, aşağıdaki ifade tablonun TTL değerini 10 gün olarak ayarlar.

print(iris[iris.sepalwidth < 2.5].persist('pyodps_iris', lifecycle=10).head(5))

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

   sepallength  sepalwidth  petallength  petalwidth             name
0          4.5         2.3          1.3         0.3      Iris-setosa
1          5.5         2.3          4.0         1.3  Iris-versicolor
2          4.9         2.4          3.3         1.0  Iris-versicolor
3          5.0         2.0          3.5         1.0  Iris-versicolor
4          6.0         2.2          4.0         1.0  Iris-versicolor

Veri kaynağı MaxCompute nesneleri içermiyor ve yalnızca pandas nesneleri içeriyorsa, persist yöntemini çağırırken MaxCompute giriş nesnesini manuel olarak belirtmeniz veya giriş nesnesini genel nesne olarak işaretlemeniz gerekir.
```
# The entrance object is o. 
# Specify the entrance object. 
df.persist('table_name', odps=o)
# Alternative operation: Mark the entrance object as a global object. 
o.to_global()
df.persist('table_name')
```

Sonuçları pandas DataFrame'e kaydetme

Sonuçları bir pandas DataFrame nesnesine kaydetmek için to_pandas yöntemini çağırabilirsiniz. wrap parametresi True olarak ayarlanırsa bir PyODPS DataFrame nesnesi döndürülür.

Örnek 1: Bir pandas DataFrame nesnesi döndürmek için to_pandas yöntemini çağırın.
```
print(type(iris[iris.sepalwidth < 2.5].to_pandas()))
```
Aşağıdaki sonuç döndürülür:
```
<class 'pandas.core.frame.DataFrame'>
```
Örnek 2: Bir PyODPS DataFrame nesnesi döndürmek için wrap parametresini True olarak ayarlayın.
```
print(type(iris[iris.sepalwidth < 2.5].to_pandas(wrap=True)))
```
Aşağıdaki sonuç döndürülür:
```
<class 'odps.df.core.DataFrame'>
```

Not

PyODPS'de open_reader yöntemini çağırabilir ve sonuçları bir pandas DataFrame nesnesine dönüştürmek için reader.to_pandas() kullanabilirsiniz. Daha fazla bilgi için bkz. Tablolar.

Çalışma zamanı parametrelerini yapılandırma

execute, persist ve to_pandas gibi hemen yürütülen yöntemler için çalışma zamanı parametrelerini yapılandırabilirsiniz. Bu ayar yalnızca MaxCompute SQL arka ucu için geçerlidir.

Genel parametreleri yapılandırın. Daha fazla bilgi için bkz. SQL.

Bu yöntemlerde hints parametresini belirtin. Bu, belirtilen çalışma zamanı parametrelerinin yalnızca geçerli hesaplama için geçerli olmasını sağlar.

print(iris[iris.sepallength < 5].to_pandas(hints={'odps.sql.mapper.split.size': 16}))

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

   sepallength  sepalwidth  petallength  petalwidth             name
0          4.5         2.3          1.3         0.3      Iris-setosa
1          4.9         2.4          3.3         1.0  Iris-versicolor

Çalışma zamanında ayrıntıları görüntüleme

Çalışma zamanında bir örneğin LogView bilgilerini görüntülemek için genel yapılandırmaları değiştirmeniz gerekir. Aşağıdaki kod bir örneği göstermektedir:

from odps import options
options.verbose = True

print(iris[iris.sepallength < 5].exclude('sepallength')[:5].execute())

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

Sql compiled:
SELECT t1.`sepalwidth`, t1.`petallength`, t1.`petalwidth`, t1.`name`
FROM odps_test_sqltask_finance.`pyodps_iris` t1
WHERE t1.`sepallength` < 5
LIMIT 5
Instance ID:
  Log view:http://logview
  
   sepalwidth  petallength  petalwidth             name
0         2.3          1.3         0.3      Iris-setosa
1         2.4          3.3         1.0  Iris-versicolor

Bir günlük kaydı işlevi belirtebilirsiniz. Aşağıdaki kod bir örneği göstermektedir:

my_logs = []
def my_logger(x):
    my_logs.append(x)
options.verbose_log = my_logger
print(iris[iris.sepallength < 5].exclude('sepallength')[:5].execute())

print(my_logs)

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

   sepalwidth  petallength  petalwidth             name
0         2.3          1.3         0.3      Iris-setosa
1         2.4          3.3         1.0  Iris-versicolor

['Sql compiled:', 'CREATE TABLE tmp_pyodps_24332bdb_4fd0_4d0d_aed4_38a443618268 LIFECYCLE 1 AS \nSELECT t1.`sepalwidth`, t1.`petallength`, t1.`petalwidth`, t1.`name` \nFROM odps_test_sqltask_finance.`pyodps_iris` t1 \nWHERE t1.`sepallength` < 5 \nLIMIT 5', 'Instance ID: 20230815034706122gbymevg*****', '  Log view:]

Collection nesnelerinin ara hesaplama sonuçlarını önbelleğe alma

DataFrame hesaplama sürecinde bazı Collection nesneleri birden çok kez kullanılır. Bir ara sürecin yürütme sonuçlarını görüntülemek için, önce hesaplamak istediğiniz bir Collection nesnesini işaretlemek üzere cache yöntemini çağırabilirsiniz. Aşağıdaki kod bir örneği göstermektedir.

Not

cache yönteminin yürütülmesi ertelenir. cache yöntemi çağrıldıktan sonra otomatik hesaplama hemen tetiklenmez.

cached = iris[iris.sepalwidth < 3.5]['sepallength', 'name'].cache()
df = cached.head(3)
print(df)

# The following result is returned:
   sepallength             name
0          4.5      Iris-setosa
1          5.5  Iris-versicolor
2          4.9  Iris-versicolor

# You can immediately retrieve the calculation result because cached is calculated. 
print(cached.head(3))

# The following result is returned:
   sepallength             name
0          4.5      Iris-setosa
1          5.5  Iris-versicolor
2          4.9  Iris-versicolor

Eşzamansız ve paralel çalıştırma

Eşzamansız çalıştırma

PyODPS DataFrame eşzamansız çalıştırmayı destekler. Aşağıdaki hemen yürütülen yöntemler için eşzamansız çalıştırmayı etkinleştirmek üzere async parametresini belirtebilirsiniz: execute, persist, head, tail ve to_pandas. timeout parametresi zaman aşımı süresini belirtir. Eşzamansız işlemler Future nesnelerini döndürür.

future = iris[iris.sepalwidth < 10].head(10, async_=True)
print(future.result())

# The following result is returned:
   sepallength  sepalwidth  petallength  petalwidth             name
0          4.5         2.3          1.3         0.3      Iris-setosa
1          5.5         2.3          4.0         1.3  Iris-versicolor
2          4.9         2.4          3.3         1.0  Iris-versicolor
3          5.0         2.0          3.5         1.0  Iris-versicolor
4          6.0         2.2          4.0         1.0  Iris-versicolor
5          6.2         2.2          4.5         1.5  Iris-versicolor
6          5.5         2.4          3.8         1.1  Iris-versicolor
7          5.5         2.4          3.7         1.0  Iris-versicolor
8          6.3         2.3          4.4         1.3  Iris-versicolor
9          5.0         2.3          3.3         1.0  Iris-versicolor

Paralel çalıştırma

Yeni tanıtılan Delay API işlemini çağırarak aşağıdaki hemen yürütülen yöntemleri erteleyebilirsiniz: execute, persist, head, tail ve to_pandas. Ardından, Future nesneleri döndürülür. Delay API işlemi çağrıldığında, sistem bağımlılığı bulur ve belirtilen eşzamanlılığa göre yöntemleri yürütür. Bu durumda eşzamansız çalıştırma desteklenir.

from odps.df import Delay
delay = Delay()  # Create a Delay object. 

df = iris[iris.sepal_width < 5].cache()  # Common dependency of subsequent expressions. 
future1 = df.sepal_width.sum().execute(delay=delay)  # Return a Future object. The execution is not started. 
future2 = df.sepal_width.mean().execute(delay=delay)
future3 = df.sepal_length.max().execute(delay=delay)
delay.execute(n_parallel=3)  # The execution starts with three concurrent threads. 
|==========================================|   1 /  1  (100.00%)        21s
print(future1.result())

# The following result is returned:
25.0
 
print(future2.result())

# The following result is returned:
2.272727272727273

Yukarıdaki örnekte, PyODPS DataFrame önce paylaşılan bağımlılığın nesnesini yürütür. Ardından, PyODPS DataFrame eşzamanlılığı 3 olarak ayarlar ve future1 ile future3 arasındaki nesneleri yürütür.

Eşzamansız çalıştırmayı etkinleştirip etkinleştirmeyeceğinizi belirtmek için delay.execute içinde async parametresini belirtebilirsiniz. Eşzamansız çalıştırma etkinleştirilirse, zaman aşımı süresini belirtmek için timeout parametresini de kullanabilirsiniz.