MaxCompute：Gunakan UDF dan pustaka Python pihak ketiga

DataFrame memungkinkan Anda menggunakan metode map pada objek Sequence untuk memanggil UDF di semua elemennya.

>>> iris.sepallength.map(lambda x: x + 1).head(5)
   sepallength
0          6.1
1          5.9
2          5.7
3          5.6
4          6.0

Jika tipe Sequence berubah setelah metode map dijalankan, Anda perlu secara eksplisit menentukan tipe baru dari Sequence.

>>> iris.sepallength.map(lambda x: 't'+str(x), 'string').head(5)
   sepallength
0         t5.1
1         t4.9
2         t4.7
3         t4.6
4         t5.0

Jika suatu fungsi mengandung closure, perubahan nilai variabel closure di luar fungsi menyebabkan nilai variabel dalam fungsi tersebut berubah.

>>> dfs = []
>>> for i in range(10):
>>>     dfs.append(df.sepal_length.map(lambda x: x + i))

Akibatnya, setiap objek SequenceExpr dalam dfs adalah df.sepal_length + 9. Untuk menyelesaikan masalah ini, Anda dapat menggunakan fungsi sebagai nilai kembalian dari fungsi lain, atau gunakan partial. Lihat dua contoh berikut.

>>> dfs = []
>>> def get_mapper(i):
>>>     return lambda x: x + i
>>> for i in range(10):
>>>     dfs.append(df.sepal_length.map(get_mapper(i)))

>>> import functools
>>> dfs = []
>>> for i in range(10):
>>>     dfs.append(df.sepal_length.map(functools.partial(lambda v, x: x + v, i)))

Contoh penggunaan counter:

from odps.udf import get_execution_context
def h(x):
    ctx = get_execution_context()
    counters = ctx.get_counters()
    counters.get_counter('df', 'add_one').increment(1)
    return x + 1
df.field.map(h)

Anda dapat menemukan nilai counter di JSONSummary dari LogView.

Untuk menggunakan UDF untuk satu baris, Anda dapat menggunakan metode apply. Parameter axis harus disetel ke 1 untuk menunjukkan bahwa operasi dilakukan pada baris. UDF dari metode apply mengambil satu parameter, yaitu satu baris data dari objek Collection sebelumnya. Anda dapat mengambil data di bidang tertentu dengan menggunakan atribut atau offset.

• Jika parameter reduce disetel ke True, objek Sequence dikembalikan. Jika tidak, objek Collection dikembalikan. Anda dapat menggunakan parameter names dan types untuk menentukan nama bidang dan jenis dari objek Sequence atau Collection yang dikembalikan. Secara default, jika Anda tidak menentukan jenis, jenis STRING digunakan.

  >>> iris.apply(lambda row: row.sepallength + row.sepalwidth, axis=1, reduce=True, types='float').rename('sepaladd').head(3)
     sepaladd
  0       8.6
  1       7.9
  2       7.9

• Jika reduce adalah False dalam UDF dari metode apply, Anda dapat menggunakan kata kunci yield untuk mengembalikan beberapa hasil baris.

  >>> iris.count()
  150
  >>>
  >>> def handle(row):
  >>>     yield row.sepallength - row.sepalwidth, row.sepallength + row.sepalwidth
  >>>     yield row.petallength - row.petalwidth, row.petallength + row.petalwidth
  >>>
  >>> iris.apply(handle, axis=1, names=['iris_add', 'iris_sub'], types=['float', 'float']).count()
  300

• Anda juga dapat memberi anotasi pada nama bidang dan jenis yang dikembalikan dalam fungsi. Anda tidak perlu menentukannya saat memanggil fungsi.

  >>> from odps.df import output
  >>>
  >>> @output(['iris_add', 'iris_sub'], ['float', 'float'])
  >>> def handle(row):
  >>>     yield row.sepallength - row.sepalwidth, row.sepallength + row.sepalwidth
  >>>     yield row.petallength - row.petalwidth, row.petallength + row.petalwidth
  >>>
  >>> iris.apply(handle, axis=1).count()
  300

• Anda juga dapat menggunakan map_reduce dari map-only, yang setara dengan metode apply dengan axis=1.

  >>> iris.map_reduce(mapper=handle).count()
  300

• Untuk menggunakan fungsi tabel yang ditentukan pengguna (UDTF) yang sudah ada di MaxCompute, tentukan nama UDTF.

  >>> iris['name', 'sepallength'].apply('your_func', axis=1, names=['name2', 'sepallength2'], types=['string', 'float'])

• Saat menggunakan metode apply untuk baris dan reduce adalah False, Anda dapat menggunakan lateral view dengan baris yang ada untuk tujuan seperti agregasi.

  >>> from odps.df import output
  >>>
  >>> @output(['iris_add', 'iris_sub'], ['float', 'float'])
  >>> def handle(row):
  >>>     yield row.sepallength - row.sepalwidth, row.sepallength + row.sepalwidth
  >>>     yield row.petallength - row.petalwidth, row.petallength + row.petalwidth
  >>>
  >>> iris[iris.category, iris.apply(handle, axis=1)]

Saat menggunakan metode apply, jika axis tidak ditentukan atau nilai dari axis adalah 0, Anda dapat memasukkan kelas agregasi kustom untuk menggabungkan semua objek Sequence.

class Agg(object):

    def buffer(self):
        return [0.0, 0]

    def __call__(self, buffer, val):
        buffer[0] += val
        buffer[1] += 1

    def merge(self, buffer, pbuffer):
        buffer[0] += pbuffer[0]
        buffer[1] += pbuffer[1]

    def getvalue(self, buffer):
        if buffer[1] == 0:
            return 0.0
        return buffer[0] / buffer[1]

>>> iris.exclude('name').apply(Agg)
   sepallength_aggregation  sepalwidth_aggregation  petallength_aggregation  petalwidth_aggregation
0                 5.843333                   3.054                 3.758667                1.198667

UDF juga dapat membaca sumber daya MaxCompute (seperti sumber daya tabel dan file) atau merujuk objek Collection sebagai sumber daya. Untuk melakukannya, Anda perlu menulis UDF sebagai closure atau kelas yang dapat dipanggil. Lihat dua contoh berikut.

>>> file_resource = o.create_resource('pyodps_iris_file', 'file', file_obj='Iris-setosa')
>>>
>>> iris_names_collection = iris.distinct('name')[:2]
>>> iris_names_collection
       sepallength
0      Iris-setosa
1  Iris-versicolor

>>> def myfunc(resources):  # resources diteruskan sesuai urutan pemanggilan
>>>     names = set()
>>>     fileobj = resources[0] # sumber daya file direpresentasikan oleh objek mirip file
>>>     for l in fileobj:
>>>         names.add(l)
>>>     collection = resources[1]
>>>     for r in collection:
>>>         names.add(r.name)  # ambil nilai dengan menggunakan nama bidang atau offset
>>>     def h(x):
>>>         if x in names:
>>>             return True
>>>         else:
>>>             return False
>>>     return h
>>>
>>> df = iris.distinct('name')
>>> df = df[df.name,
>>>         df.name.map(myfunc, resources=[file_resource, iris_names_collection], rtype='boolean').rename('isin')]
>>>
>>> df
              name   isin
0      Iris-setosa   True
1  Iris-versicolor   True
2   Iris-virginica  False

Jika axis adalah 1 dalam operasi baris, Anda perlu menulis penutup fungsi atau kelas yang dapat dipanggil. Untuk operasi agregasi pada kolom, Anda hanya perlu menggunakan fungsi __init__ untuk membaca sumber daya.

>>> words_df
                     sentence
0                 Hello World
1                Hello Python
2  Life is short I use Python
>>>
>>> import pandas as pd
>>> stop_words = DataFrame(pd.DataFrame({'stops': ['is', 'a', 'I']}))
>>>
>>> @output(['sentence'], ['string'])
>>> def filter_stops(resources):
>>>     stop_words = set([r[0] for r in resources[0]])
>>>     def h(row):
>>>         return ' '.join(w for w in row[0].split() if w not in stop_words),
>>>     return h
>>>
>>> words_df.apply(filter_stops, axis=1, resources=[stop_words])
                sentence
0            Hello World
1           Hello Python
2  Life short use Python