MaxCompute:Kasus penggunaan

Tulis skrip UDF Python dan simpan sebagai file sum_pandas.py. Kode berikut memberikan contohnya:

from odps.udf import annotate
import pandas as pd

@annotate("string, string -> string")
class SumColumns(object):
    def evaluate(self, arg1, arg2):
        # Konversi argumen input menjadi DataFrame pandas.
        df = pd.DataFrame({'col1': arg1.split(','), 'col2': arg2.split(',')})

        # Proses data dengan menggunakan pandas.
        # Contoh ini menghitung jumlah dari dua kolom.
        df['sum'] = df['col1'].astype(int) + df['col2'].astype(int)

        # Konversi hasil menjadi string dan kembalikan.
        result = ','.join(df['sum'].astype(str).values)
        return result

Unggah skrip sum_pandas.py sebagai resource ke proyek MaxCompute Anda. Untuk informasi selengkapnya, lihat Add resources. Perintah contoh berikut menunjukkan cara mengunggah skrip tersebut:

ADD PY sum_pandas.py -f;

Daftarkan skrip sum_pandas.py yang telah Anda unggah sebagai fungsi user-defined SumColumns. Untuk informasi selengkapnya, lihat Register a function. Perintah contoh berikut menunjukkan cara mendaftarkan fungsi tersebut:

CREATE FUNCTION SumColumns AS 'sum_pandas.SumColumns' USING 'sum_pandas.py';

Buat tabel uji bernama testsum dan masukkan data uji ke dalam tabel tersebut.

CREATE TABLE testsum (col1 string, col2 string);
INSERT INTO testsum VALUES ('1,2,3','1,2,3'),('1,2,3','3,2,1'),('1,2,3','4,5,6');

Panggil UDF dan tentukan image yang ada dengan menggunakan flag.

set odps.sql.python.version=cp37;
set odps.session.image = <image_name>;
SELECT SumColumns(col1,col2) AS result FROM testsum;

Hasil berikut dikembalikan:

+------------+
| result     |
+------------+
| 2,4,6      |
| 4,4,4      |
| 5,7,9      |
+------------+

Buat tabel uji bernama test_float_col dan masukkan data uji ke dalam tabel tersebut.

CREATE TABLE test_float_col (col1 double);
INSERT INTO test_float_col VALUES (3.75),(2.51);

Tulis skrip PyODPS untuk menghitung nilai psi(col1), simpan skrip tersebut sebagai file psi_col.py, lalu jalankan file tersebut. Kode berikut memberikan contohnya:

import os
from odps import ODPS, options

def my_psi(v):
    from scipy.special import psi

    return float(psi(v))

# Jika proyek memiliki isolasi yang diaktifkan, opsi berikut tidak diperlukan.
options.sql.settings = {"odps.isolation.session.enable": True}

o = ODPS(
      # Pastikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID diatur ke ID AccessKey Anda,
      # dan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_SECRET diatur ke rahasia AccessKey Anda.
      # Kami menyarankan agar Anda tidak menyematkan ID AccessKey dan rahasia AccessKey secara langsung dalam kode.
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
      project='your-default-project',
      endpoint='your-end-point'
)

df = o.get_table("test_float_col").to_df()
# Jalankan pekerjaan dan ambil hasilnya.
df.col1.map(my_psi).execute(image='scipy')
# Simpan hasilnya ke tabel lain.
df.col1.map(my_psi).persist("result_table", image='scipy')

Lihat hasilnya di tabel result_table.

SELECT * FROM result_table

Hasil berikut dikembalikan:

+------------+
| col1       |
+------------+
| 1.1825373886117962 |
| 0.7080484451910534 |
+------------+

Buat tabel uji bernama test_float_col dan masukkan data uji ke dalam tabel tersebut.

CREATE TABLE test_float_col (col1 double);
INSERT INTO test_float_col VALUES (3.75),(2.51);

Tulis skrip MaxFrame untuk menghitung nilai psi(col1), simpan skrip tersebut sebagai file psi_col.py, lalu jalankan file tersebut. Kode berikut memberikan contohnya:

import os
from odps import ODPS, options
from maxframe.session import new_session
import maxframe.dataframe as md


from maxframe.config import options
from maxframe import config

# Rujuk image SciPy bawaan.
config.options.sql.settings = {
    "odps.session.image": "scipy"
}
def my_psi(v):
    from scipy.special import psi
    return float(psi(v))

o = ODPS(
      # Pastikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID diatur ke ID AccessKey Anda,
      # dan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_SECRET diatur ke rahasia AccessKey Anda.
      # Kami menyarankan agar Anda tidak menyematkan ID AccessKey dan rahasia AccessKey secara langsung dalam kode.
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
      project='your-default-project',
      endpoint='your-end-point'
)

# Buat sesi MaxFrame.
session = new_session(o)
df = md.read_odps_table('test_float_col')

# Jalankan pekerjaan dan ambil hasilnya.
print(df.col1.map(my_psi).execute().fetch())

Hasil berikut dikembalikan:

0    1.182537
1    0.708048
Name: col1, dtype: float64