Platform For AI：Platform fitur dan rekayasa fitur

Definisi dan pelaksanaan transformasi fitur statistik tunggal

1. Definisikan nama tabel input dan output.

   input_agg_table_name = "rec_sln_demo_behavior_table_preprocess_wide_v1"
   ds_agg = MaxComputeDataSource(table=input_agg_table_name, project=project)
   output_agg_table_name = "rec_sln_demo_behavior_test_agg_v1"

2. Definisikan fitur statistik.

   feature_agg1 = Feature(
               name="user_avg_praise_count_1d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="avg", # Nama fungsi agregat. Nilai opsional adalah 'avg', 'sum', 'min', 'max'.
                   condition=Condition(field="event", value="expr", operator="<>"), # Tetapkan kondisi yang terpenuhi ketika nilai bidang "event" tidak sama dengan "expr". Anda dapat memahami logika spesifik dari pernyataan SQL yang dihasilkan.
                   group_by_keys="user_id", # Kunci yang sesuai dengan group by.
                   window_size=DayOf(1), # Ukuran jendela, yang satu hari di sini.
               ),
           )

3. Anda dapat membuat pipeline dan menjalankan transformasi fitur statistik.

   agg_pipeline = fs_client.create_pipeline(ds_agg, output_agg_table_name).add_feature_transform([feature_agg1])

4. Hasilkan dan jalankan transformasi.

   execute_date = '20240605'
   print("transform_info = ", agg_pipeline.transform_info)
   output_agg_table = agg_pipeline.execute(execute_date, drop_table=True)

5. Lihat isi transform_info.

   transform_info_agg = output_agg_table.transform_info
   transform_info_agg

6. Lihat hasilnya.

   pd_ret = output_agg_table.to_pandas(execute_date, limit=20)
   pd_ret

Gabungan otomatis untuk transformasi fitur statistik dengan jendela berbeda

1. Definisikan nama tabel output.

   output_agg_table_name_2 = "rec_sln_demo_behavior_test_agg_v2"

2. Definisikan fitur statistik untuk beberapa jendela berbeda. Ukuran jendela yang didefinisikan adalah 1, 3, 7, 15, dan 30 hari.

   feature_agg1 = Feature(
               name="user_avg_praise_count_1d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="avg",
                   condition=Condition(field="event", value="expr", operator="<>"),
                   group_by_keys="user_id",
                   window_size=DayOf(1),
               ),
           )
   feature_agg2 = Feature(
               name="user_avg_praise_count_3d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="avg",
                   condition=Condition(field="event", value="expr", operator="<>"),
                   group_by_keys="user_id",
                   window_size=DayOf(3),
               ),
           )
   feature_agg3 = Feature(
               name="user_avg_praise_count_7d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="avg",
                   condition=Condition(field="event", value="expr", operator="<>"),
                   group_by_keys="user_id",
                   window_size=DayOf(7),
               ),
           )
   feature_agg4 = Feature(
               name="user_avg_praise_count_15d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="avg",
                   condition=Condition(field="event", value="expr", operator="<>"),
                   group_by_keys="user_id",
                   window_size=DayOf(15),
               ),
           )
   feature_agg5 = Feature(
               name="user_avg_praise_count_30d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="avg",
                   condition=Condition(field="event", value="expr", operator="<>"),
                   group_by_keys="user_id",
                   window_size=DayOf(30),
               ),
           )

3. Buat pipeline.

   agg_pipeline_2 = fs_client.create_pipeline(ds_agg, output_agg_table_name_2).add_feature_transform([feature_agg1, feature_agg2, feature_agg3, feature_agg4, feature_agg5])

4. Hasilkan dan jalankan pipeline.

   execute_date = '20240605'
   output_agg_table_2 = agg_pipeline_2.execute(execute_date, drop_table=True)

5. Lihat hasil transformasi.

   transform_info_agg_2 = output_agg_table_2.transform_info
   transform_info_agg_2

6. Lihat hasil run tabel.

   pd_ret_2 = output_agg_table_2.to_pandas(execute_date, limit=20)
   pd_ret_2

Penggabungan otomatis dan derivasi tipe untuk beberapa proses transformasi fitur statistik

Untuk mengoptimalkan perhitungan, beberapa fitur dengan ukuran jendela yang sama digabungkan dan dihitung secara otomatis dalam blok jendela grup yang sama. Proses perhitungan melibatkan perubahan tipe. Sebagai contoh, `avg` mengubah tipe `bigint` menjadi tipe `double`. Sulit untuk mengingat banyak tipe fitur input. Oleh karena itu, proses transformasi fitur statistik mendukung derivasi tipe otomatis, yang berarti Anda tidak perlu menentukan tipe terlebih dahulu. Tipe fitur yang dihasilkan secara otomatis diturunkan selama definisi fitur.

1. Definisikan nama tabel output.

   output_agg_table_name_3 = "rec_sln_demo_behavior_test_agg_v3"

2. Definisikan lebih banyak fitur dari tipe berbeda.

   feature_agg6 = Feature(
               name="user_expr_cnt_1d",
               transform=AggregationTransform(
                   agg_func="count",
                   condition=Condition(field="event", value="expr", operator="="),
                   group_by_keys="user_id",
                   window_size=DayOf(1),
               )
           )
   feature_agg7 = Feature(
               name="user_expr_item_id_dcnt_1d",
               input=['item_id'],
               transform=AggregationTransform(
                   agg_func="count",
                   condition=Condition(field="event", value="expr", operator="="),
                   group_by_keys="user_id",
                   window_size=DayOf(1),
               ),
           )
   feature_agg8 = Feature(
               name="user_sum_praise_count_1d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="sum",
                   condition=Condition(field="event", value="expr", operator="<>"),
                   group_by_keys="user_id",
                   window_size=DayOf(1),
               ),
           )
   feature_agg9 = Feature(
               name="user_sum_praise_count_3d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="sum",
                   condition=Condition(field="event", value="expr", operator="<>"),
                   group_by_keys="user_id",
                   window_size=DayOf(3),
               ),
           )

3. Buat pipeline.

   agg_pipeline_3 = fs_client.create_pipeline(ds_agg, output_agg_table_name_3).add_feature_transform([feature_agg1, feature_agg2, feature_agg3, feature_agg4, feature_agg5, feature_agg6, feature_agg7, feature_agg8, feature_agg9])

4. Hasilkan dan jalankan pipeline.

   execute_date = '20240605'
   output_agg_table_3 = agg_pipeline_3.execute(execute_date, drop_table=True)

5. Lihat hasil transformasi.

   transform_info_agg_3 = output_agg_table_3.transform_info
   transform_info_agg_3

6. Lihat hasil run tabel.

   pd_ret_3 = output_agg_table_3.to_pandas(execute_date, limit=20)
   pd_ret_3

Fungsi ekstensi otomatis bawaan untuk mendukung ekstensi otomatis transformasi fitur statistik

Mengimplementasikan setiap fitur statistik secara manual kompleks karena jumlah fitur yang besar, termasuk ukuran jendela yang berbeda dan banyak kombinasi perhitungan fungsi agregat. Sistem menyediakan fungsi ekstensi otomatis bawaan. Anda hanya perlu menentukan fitur input yang akan dihitung, dan sistem secara otomatis menghasilkan dan menyelesaikan definisi untuk ratusan fitur statistik.

1. Tentukan fitur input yang akan dihitung.

   name_prefix = "user_"
   input_list = ["playtime", "duration", "click_count", "praise_count"]
   event_name = 'event'
   event_type = 'expr'
   group_by_key = "user_id"
   count_feature_list = auto_count_feature_transform(name_prefix, input_list, event_name, event_type, group_by_key)
   print("len_count_feature_list = ", len(count_feature_list))
   print("count_feature_list = ", count_feature_list)

2. Definisikan nama tabel output dan buat pipeline.

   output_agg_table_name_4 = "rec_sln_demo_behavior_test_agg_v4"
   agg_pipeline_4 =fs_client.create_pipeline(ds_agg, output_agg_table_name_4).add_feature_transform(count_feature_list)

3. Hasilkan dan jalankan pipeline.

   execute_date = '20240605'
   output_agg_table_4 = agg_pipeline_4.execute(execute_date, drop_table=True)

4. Lihat hasil transformasi.

   transform_info_agg_4 = output_agg_table_4.transform_info
   transform_info_agg_4

5. Lihat hasil run tabel.

   pd_ret_4 = output_agg_table_4.to_pandas(execute_date, limit=20)
   pd_ret_4

Dukungan untuk transformasi berbeda pada kunci grup yang berbeda

Bagian sebelumnya menjelaskan metode pemrosesan ketika semua kunci grup sama. Sistem juga mendukung operasi transformasi pada kunci grup yang berbeda. Berikut adalah contohnya:

1. Definisikan nama tabel output.

   input_agg_table_name = "rec_sln_demo_behavior_table_preprocess_wide_v1"
   ds_agg = MaxComputeDataSource(table=input_agg_table_name, project=project)
   output_agg_table_name_5 = "rec_sln_demo_behavior_test_agg_v5"

2. Definisikan fitur untuk kunci grup yang berbeda.

   feature_agg1 = Feature(
               name="item__sum_follow_cnt_15d",
               input=['follow_cnt'],
               transform=AggregationTransform(
                   agg_func="sum",
                   condition=Condition(field="event", value="expr", operator="="),
                   group_by_keys="item_id",
                   window_size=DayOf(1),
               )
           )
   feature_agg2 = Feature(
               name="author__max_follow_cnt_15d",
               input=['follow_cnt'],
               transform=AggregationTransform(
                   agg_func="max",
                   condition=Condition(field="event", value="expr", operator="="),
                   group_by_keys="author",
                   window_size=DayOf(15),
               ),
           )

3. Buat pipeline.

   agg_pipeline_5 = fs_client.create_pipeline(ds_agg, output_agg_table_name_5).add_feature_transform([feature_agg1, feature_agg2])

4. Hasilkan dan jalankan pipeline.

   execute_date = '20240605'
   output_agg_table_5 = agg_pipeline_5.execute(execute_date, drop_table=True)

5. Lihat hasil transformasi.

   transform_info_agg_5 = output_agg_table_5.transform_info
   transform_info_agg_5

Proses perhitungan fungsi agregat sederhana

Fungsi agregat sederhana mencakup count, sum, max, dan min. Proses perhitungan untuk fungsi agregat ini relatif sederhana. Setelah ringkasan tingkat harian, ringkasan lebih lanjut selama beberapa hari dilakukan untuk mendapatkan hasil akhir. Selain itu, bagian ini juga memperkenalkan tabel antara harian dan penggunaan Fungsi yang Ditentukan Pengguna (UDF) (Ikhtisar UDF MaxCompute) untuk mendapatkan hasil perhitungan akhir. Dalam proses perhitungan aktual, proses untuk pengguna menjalankan rekayasa fitur sama dengan transformasi fitur konvensional yang dijelaskan sebelumnya. SDK Python FeatureStore secara otomatis menangani pembuatan tabel antara harian, pembuatan UDF, pengunggahan sumber daya, dan pendaftaran fungsi otomatis. Anda dapat menerapkan proses rekayasa fitur tanpa menyadari detail-detail ini.

1. Definisikan nama tabel input dan output.

   input_window_table_name = "rec_sln_demo_behavior_table_preprocess_wide_v1"
   ds_window_1 = MaxComputeDataSource(table=input_agg_table_name, project=project)
   output_window_table_name_1 = "rec_sln_demo_behavior_test_window_v1"

2. Definisikan fitur WindowTransform.

   win_feature1 = Feature(
       name="item__kv_gender_click_7d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="count",
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="item_id",
           window_size=DayOf(7),
       ),
   )
   win_feature2 = Feature(
       name="item__kv_gender_click_cnt_7d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="sum", # Fungsi agregat. Nilai opsional adalah 'sum', 'avg', 'max', 'min'.
           agg_field="click_count", # Lakukan perhitungan fungsi agregat pada fitur ini.
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="item_id",
           window_size=DayOf(7),
       ),
   )

3. Buat pipeline dan jalankan transformasi fitur WindowTransform.

   window_pipeline_1 = fs_client.create_pipeline(ds_window_1, output_window_table_name_1).add_feature_transform([win_feature1, win_feature2], keep_input_columns=True)

4. Hasilkan dan jalankan transformasi.

   Proses generasi ini membuat tabel antara sementara harian. Pada titik ini, `DROP TABLE` hanya menghapus hasil akhir dan tidak menghapus tabel antara sementara.

   execute_date = '20240605'
   print("transform_info = ", window_pipeline_1.transform_info)
   output_window_table_1 = window_pipeline_1.execute(execute_date, drop_table=True)

   Selain itu, karena melibatkan statistik untuk beberapa hari (misalnya, contoh sebelumnya menghitung data selama 7 hari), tabel antara sementara biasanya hanya menghitung data untuk partisi terbaru. Untuk alasan ini, sistem menyediakan parameter `backfill_partitions`. Saat Anda menjalankan proses untuk pertama kali, Anda dapat mengatur parameter ini ke `True`, dan sistem secara otomatis mengisi ulang data untuk hari-hari yang bergantung. Misalnya, jika penghitungan melibatkan data selama 7 hari, sistem secara otomatis menyelesaikan 7 hari data tersebut. Untuk rutinitas selanjutnya, Anda dapat mengatur parameter ke `False` untuk menyelesaikan data hanya untuk partisi hari terbaru.

   execute_date = '20240506'
   output_window_table_1 = window_pipeline_1.execute(execute_date, backfill_partitions=True)

   • Ketika parameter `backfill_partitions` diatur ke `True`, sistem secara otomatis menyelesaikan data untuk hari-hari yang bergantung dari tabel antara sementara. Kami sarankan Anda melakukan ini selama rutinitas pertama kali.

   • Jika jumlah hari yang harus dihitung besar, menjalankan kode sebelumnya membutuhkan waktu lama.

5. Lihat data tabel sink.

   window_ret_1 = output_window_table_1.to_pandas(execute_date, limit=50)
   window_ret_1

6. Lihat proses perhitungan aktual.

   window_pipeline_1.transform_info

   Seperti yang bisa Anda lihat dari proses perhitungan, sistem menghasilkan tabel antara sementara harian bernama `rec_sln_demo_behavior_table_preprocess_wide_v1_tmp_daily`. Tabel ini merangkum hasil harian dan menyimpannya di partisi tetap, yang menghindari perhitungan berulang.

   Selain itu, UDF bernama `count_kv` digunakan untuk menghitung hasil akhir. UDF ini secara otomatis mengklasifikasikan dan merangkum hasil statistik ke dalam peta hasil, yang disimpan dalam format string. Ini memudahkan inferensi hasil offline dan online selanjutnya.

Konten sebelumnya memperkenalkan proses perhitungan untuk fungsi agregat sederhana, menggunakan `count` dan `sum` sebagai contoh. Meskipun proses ini melibatkan konsep seperti tabel antara sementara harian dan UDF, alur inti sama dengan operasi transformasi data konvensional dan tidak menambah kompleksitas operasional. Fungsi agregat sederhana lainnya, seperti `max` dan `min`, serupa.

Proses perhitungan fungsi agregat avg

Karena merata-ratakan hasil rata-rata harian mengarah pada perhitungan yang tidak akurat, fungsi agregat avg memiliki proses perhitungan khusus. Metode yang benar adalah pertama-tama menghitung total sum (sum_v) dan total count (count_v) untuk seluruh periode, kemudian menghitung rata-rata menggunakan rumus sum_v/count_v.

Meskipun fungsi agregat ini didokumentasikan secara terpisah, detail perhitungan kompleksnya dienkapsulasi dalam transform_info. Oleh karena itu, Anda tidak perlu memahami detail dasarnya. Anda dapat menggunakan fungsi ini seperti fitur konvensional untuk menghasilkan hasil akhir.

1. Definisikan nama tabel input dan output.

   input_window_table_name = "rec_sln_demo_behavior_table_preprocess_wide_v1"
   ds_window_1 = MaxComputeDataSource(table=input_window_table_name, project=project)
   output_window_table_name_2 = "rec_sln_demo_behavior_test_window_v2"

2. Anda dapat mendefinisikan fitur WindowTransform.

   win_feature1 = Feature(
       name="item__kv_gender_click_avg_7d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="avg",
           agg_field="click_count",
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="item_id",
           window_size=DayOf(7),
       ),
   )
   win_feature2 = Feature(
       name="item__kv_gender_click_avg_15d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="avg",
           agg_field="click_count",
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="item_id",
           window_size=DayOf(15),
       ),
   )

3. Buat pipeline dan jalankan transformasi fitur WindowTransform.

   window_pipeline_2 = fs_client.create_pipeline(ds_window_1, output_window_table_name_2).add_feature_transform([win_feature1, win_feature2])

4. Hasilkan dan jalankan transformasi.

   execute_date = '20240605'
   print("transform_info = ", window_pipeline_2.transform_info)
   output_window_table_2 = window_pipeline_2.execute(execute_date, drop_table=True)

5. Lihat data tabel sink.

   window_ret_2 = output_window_table_2.to_pandas(execute_date, limit=50)
   window_ret_2

Proses perhitungan fungsi untuk beberapa kunci grup

Demikian pula, WindowTransform mendukung perhitungan simultan dengan beberapa kunci grup. Hasilnya kemudian digabungkan ke tabel input. Berikut adalah contohnya:

1. Definisikan nama tabel input dan output.

   input_window_table_name = "rec_sln_demo_behavior_table_preprocess_wide_v1"
   ds_window_1 = MaxComputeDataSource(table=input_window_table_name, project=project)
   output_window_table_name_3 = "rec_sln_demo_behavior_test_window_v3"

2. Definisikan fitur WindowTransform.

   win_feature1 = Feature(
       name="item__kv_gender_click_7d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="count",
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="item_id",
           window_size=DayOf(7),
       ),
   )
   win_feature2 = Feature(
       name="item__kv_gender_click_cnt_7d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="sum",
           agg_field="click_count",
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="item_id",
           window_size=DayOf(7),
       ),
   )
   win_feature3 = Feature(
       name="author__kv_gender_click_15d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="count",
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="author",
           window_size=DayOf(7),
       ),
   )
   win_feature4 = Feature(
       name="author__kv_gender_click_cnt_15d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="sum",
           agg_field="click_count",
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="author",
           window_size=DayOf(7),
       ),
   )

3. Buat pipeline dan jalankan transformasi fitur WindowTransform.

   window_pipeline_3 = fs_client.create_pipeline(ds_window_1, output_window_table_name_3).add_feature_transform([win_feature1, win_feature2, win_feature3, win_feature4])

4. Hasilkan dan jalankan transformasi.

   execute_date = '20240605'
   print("transform_info = ", window_pipeline_3.transform_info)
   output_window_table_3 = window_pipeline_3.execute(execute_date, drop_table=True)

5. Lihat data tabel sink.

   window_ret_3 = output_window_table_3.to_pandas(execute_date, limit=50)
   window_ret_3

Fungsi ekstensi otomatis bawaan untuk mendukung ekstensi otomatis fitur WindowTransform

Menyerupai transformasi fitur statistik, mengimplementasikan setiap fitur statistik WindowTransform secara manual kompleks karena jumlah fitur yang besar, termasuk ukuran jendela yang berbeda dan banyak kombinasi perhitungan fungsi agregat. Sistem menyediakan fungsi ekstensi otomatis bawaan. Anda hanya perlu menentukan fitur input yang akan dihitung, dan sistem secara otomatis menghasilkan dan menyelesaikan definisi untuk ratusan fitur statistik.

1. Tentukan fitur input yang akan dihitung.

   name_prefix = "item"
   input_list = ['gender']
   agg_field = ["click_count"]
   event_name = 'event'
   event_type = 'click'
   group_by_key = "item_id"
   window_size = [7, 15, 30, 45]
   window_transform_feature_list = auto_window_feature_transform(name_prefix, input_list, agg_field, event_name, event_type, group_by_key, window_size)
   print("len_window_transform_feature_list = ", len(window_transform_feature_list))
   print("window_transform_feature_list = ", window_transform_feature_list)

2. Definisikan nama tabel output dan buat pipeline.

   input_window_table_name = "rec_sln_demo_behavior_table_preprocess_wide_v1"
   ds_window_1 = MaxComputeDataSource(table=input_window_table_name, project=project)
   output_window_table_name_4 = "rec_sln_demo_behavior_test_window_v4"
   window_pipeline_4 =fs_client.create_pipeline(ds_window_1, output_window_table_name_4).add_feature_transform(window_transform_feature_list)

3. Hasilkan dan jalankan transformasi.

   execute_date = '20240605'
   print("transform_info = ", window_pipeline_4.transform_info)
   output_window_table_4 = window_pipeline_4.execute(execute_date, drop_table=True)

Transformasi JoinTransform

Dalam proses rekayasa fitur sebelumnya, terutama untuk AggregationTransform dan WindowTransform, inputnya adalah tabel perilaku, dan hasil output juga disimpan dalam tabel perilaku. Namun, dalam kebanyakan kasus, tabel akhir yang dibutuhkan untuk penggunaan online bukanlah tabel perilaku, tetapi tabel fitur yang dihasilkan dengan menggabungkan tabel perilaku dengan tabel lain, seperti tabel pengguna atau tabel item.

Oleh karena itu, `JoinTransform` diperkenalkan untuk mendukung penggabungan fitur dari AggregationTransform dan WindowTransform dengan tabel pengguna atau item yang ada.

Hubungkan JoinTransform dengan WindowTransform

1. Definisikan tabel input WindowTransform.

   window_table_name = 'rec_sln_demo_behavior_table_preprocess_wide_v1'
   ds_window_1 = MaxComputeDataSource(table=window_table_name, project=project)

2. Definisikan fitur WindowTransform.

   win_fea1 = Feature(
       name="item__kv_gender_click_7d",
       input=["gender"],
       transform=WindowTransform(
           agg_func="count",
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="item_id",
           window_size=DayOf(7),
     )
   )

3. Buat pipeline.

   Catatan

   Karena tabel lain perlu digabungkan nanti, tabel output tidak ditentukan di sini.

   win_pipeline_1 = fs_client.create_pipeline(ds_window_1).add_feature_transform([win_fea1])

4. Definisikan tabel input dan output JoinTransform.

   item_table_name = 'rec_sln_demo_item_table_preprocess_v1'
   ds_join_1 = MaxComputeDataSource(table=item_table_name, project=project)
   output_table_name = 'rec_sln_demo_item_table_v1_fs_window_debug_v1'

5. Buat pipeline JoinTransform dan hubungkannya dengan pipeline WindowTransform.

   join_pipeline_1 = fs_client.create_pipeline(ds_join_1, output_table_name).merge(win_pipeline_1)

6. Hasilkan dan jalankan transformasi.

   execute_date = '20240605'
   output_join_table_1 = join_pipeline_1.execute(execute_date, drop_table=True)

7. Lihat data tabel sink.

   join_ret_1 = output_join_table_1.to_pandas(execute_date, limit = 50)
   join_ret_1

8. Lihat proses perhitungan aktual.

   output_join_table_1.transform_info

Hubungkan JoinTransform dengan AggregationTransform

1. Definisikan tabel input AggregationTransform.

   agg_table_name = 'rec_sln_demo_behavior_table_preprocess_wide_v1'
   ds_agg_1 = MaxComputeDataSource(table=agg_table_name, project=project)

2. Definisikan fitur AggregationTransform.

   agg_fea1 = Feature(
       name="user_avg_praise_count_1d",
       input=["praise_count"],
       transform=AggregationTransform(
           agg_func="avg",
           condition=Condition(field="event", value="expr", operator="<>"),
           group_by_keys="user_id",
           window_size=DayOf(1),
       ),
   )

3. Buat pipeline.

   Catatan

   Karena tabel lain perlu digabungkan nanti, tabel output tidak ditentukan di sini.

   agg_pipeline_1 = fs_client.create_pipeline(ds_agg_1).add_feature_transform([agg_fea1])

4. Definisikan tabel input dan output JoinTransform.

   user_table_name = 'rec_sln_demo_user_table_preprocess_v1'
   ds_join_2 = MaxComputeDataSource(table=user_table_name, project=project)
   output_table_name_2 = 'rec_sln_demo_user_table_v1_fs_window_debug_v1'

5. Buat pipeline JoinTransform dan hubungkannya dengan pipeline AggregationTransform.

   join_pipeline_2 = fs_client.create_pipeline(ds_join_2, output_table_name_2).merge(agg_pipeline_1, keep_input_columns=False)

6. Hasilkan dan jalankan transformasi.

   execute_date = '20240605'
   output_join_table_2 = join_pipeline_2.execute(execute_date, drop_table=True)

7. Lihat data tabel sink.

   join_ret_2 = output_join_table_2.to_pandas(execute_date, limit = 50)
   join_ret_2

8. Lihat proses perhitungan aktual.

   output_join_table_2.transform_info