Platform For AI:Feature platform and feature engineering

Définition et exécution d'une seule transformation de fonctionnalité statistique

1. Définissez les noms des tables d'entrée et de sortie.

   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. Définissez une fonctionnalité statistique.

   feature_agg1 = Feature(
               name="user_avg_praise_count_1d",
               input=["praise_count"],
               transform=AggregationTransform(
                   agg_func="avg", # Name of the aggregate function. Optional values are 'avg', 'sum', 'min', 'max'.
                   condition=Condition(field="event", value="expr", operator="<>"), # Condition: the "event" field is not equal to "expr".
                   group_by_keys="user_id", # The key corresponding to group by.
                   window_size=DayOf(1), # The window size, which is 1 day here.
               ),
           )

3. Créez un pipeline et exécutez la transformation de fonctionnalité statistique.

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

4. Générez et exécutez la transformation.

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

5. Consultez le contenu de transform_info.

   transform_info_agg = output_agg_table.transform_info
   transform_info_agg

6. Consultez les résultats.

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

Jointure automatique pour les transformations de fonctionnalités statistiques avec différentes fenêtres

1. Définissez le nom de la table de sortie.

   output_agg_table_name_2 = "rec_sln_demo_behavior_test_agg_v2"

2. Définissez des fonctionnalités statistiques pour plusieurs fenêtres différentes. Les tailles de fenêtre définies sont de 1, 3, 7, 15 et 30 jours.

   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. Créez un 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. Générez et exécutez le pipeline.

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

5. Consultez les résultats de la transformation.

   transform_info_agg_2 = output_agg_table_2.transform_info
   transform_info_agg_2

6. Consultez les résultats de l'exécution de la table.

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

Fusion automatique et dérivation de type pour plusieurs processus de transformation de fonctionnalités statistiques

Pour optimiser les calculs, plusieurs fonctionnalités ayant la même taille de fenêtre sont automatiquement fusionnées et calculées dans le même bloc de fenêtre de groupe. Le processus de calcul implique des changements de type. Par exemple, avg transforme le type bigint en type double. Comme il est difficile de se souvenir des types de toutes les fonctionnalités d'entrée, le processus de transformation de fonctionnalité statistique prend en charge la dérivation automatique de type. Le type de la fonctionnalité résultante est automatiquement dérivé lors de la définition de la fonctionnalité.

1. Définissez le nom de la table de sortie.

   output_agg_table_name_3 = "rec_sln_demo_behavior_test_agg_v3"

2. Définissez davantage de fonctionnalités de différents types.

   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. Créez un 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. Générez et exécutez le pipeline.

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

5. Consultez les résultats de la transformation.

   transform_info_agg_3 = output_agg_table_3.transform_info
   transform_info_agg_3

6. Consultez les résultats de l'exécution de la table.

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

Fonctions d'extension automatique intégrées pour prendre en charge l'extension automatique des transformations de fonctionnalités statistiques

La mise en œuvre manuelle de chaque fonctionnalité statistique est complexe en raison du grand nombre de fonctionnalités, qui incluent différentes tailles de fenêtre et de nombreuses combinaisons de fonctions d'agrégation. Le système fournit des fonctions d'extension automatique intégrées. Spécifiez les fonctionnalités d'entrée à compter, et le système génère automatiquement les définitions de centaines de fonctionnalités statistiques.

1. Spécifiez les fonctionnalités d'entrée à compter.

   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. Définissez le nom de la table de sortie et créez un 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. Générez et exécutez le pipeline.

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

4. Consultez les résultats de la transformation.

   transform_info_agg_4 = output_agg_table_4.transform_info
   transform_info_agg_4

5. Consultez les résultats de l'exécution de la table.

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

Prise en charge de la transformation simultanée de différentes clés de groupe

Les sections précédentes décrivent la méthode de traitement lorsque toutes les clés de groupe sont identiques. Le système prend également en charge les opérations de transformation sur différentes clés de groupe. Voici un exemple :

1. Définissez le nom de la table de sortie.

   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. Définissez des fonctionnalités pour différentes clés de groupe.

   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. Créez un pipeline.

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

4. Générez et exécutez le pipeline.

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

5. Consultez les résultats de la transformation.

   transform_info_agg_5 = output_agg_table_5.transform_info
   transform_info_agg_5

Processus de calcul de fonction d'agrégation simple

Les fonctions d'agrégation simples incluent count, sum, max et min. Le processus de calcul de ces fonctions d'agrégation est relativement simple. Après un résumé au niveau quotidien, un autre résumé sur plusieurs jours est effectué pour obtenir le résultat final. Cette section présente également les tables intermédiaires quotidiennes et les fonctions définies par l'utilisateur (UDF) (Aperçu des UDF MaxCompute) pour obtenir le résultat de calcul final. Dans le processus de calcul réel, l'exécution de l'ingénierie des fonctionnalités est identique à la transformation de fonctionnalité conventionnelle décrite précédemment. Le SDK Python FeatureStore gère automatiquement la création de tables intermédiaires quotidiennes, la génération d'UDF, le téléchargement des ressources et l'enregistrement automatique des fonctions.

1. Définissez les noms des tables d'entrée et de sortie.

   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. Définissez les fonctionnalités 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", # Aggregate function. Optional values are 'sum', 'avg', 'max', 'min'.
           agg_field="click_count", # Perform the aggregate function calculation on this feature.
           condition=Condition(field="event", value="click", operator="="),
           group_by_keys="item_id",
           window_size=DayOf(7),
       ),
   )

3. Créez un pipeline et exécutez la transformation de fonctionnalité 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. Générez et exécutez la transformation.

   Ce processus de génération crée une table intermédiaire temporaire quotidienne. À ce stade, DROP TABLE supprime uniquement le résultat final et ne supprime pas la table temporaire intermédiaire.

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

   De plus, comme les statistiques portent sur plusieurs jours (par exemple, l'exemple précédent compte les données sur 7 jours), la table temporaire intermédiaire calcule généralement les données uniquement pour la dernière partition. Pour cette raison, le système fournit le paramètre backfill_partitions. Lors de la première exécution du processus, vous pouvez définir ce paramètre sur True, et le système remplit automatiquement les données pour les jours dépendants. Par exemple, si le comptage implique 7 jours de données, le système complète automatiquement les 7 jours de données. Pour les exécutions routinières ultérieures, vous pouvez définir le paramètre sur False pour compléter les données uniquement pour la partition du dernier jour.

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

   • Lorsque le paramètre backfill_partitions est défini sur True, le système complète automatiquement les données pour les jours dépendants de la table intermédiaire temporaire. Effectuez cette opération lors de la première exécution routinière.

   • Si le nombre de jours à compter est élevé, l'exécution du code précédent prend beaucoup de temps.

5. Consultez les données de la table de destination.

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

6. Consultez le processus de calcul réel.

   window_pipeline_1.transform_info

   Comme vous pouvez le voir dans le processus de calcul, le système génère une table temporaire intermédiaire quotidienne nommée rec_sln_demo_behavior_table_preprocess_wide_v1_tmp_daily. Cette table résume les résultats quotidiens et les stocke dans une partition fixe, ce qui évite les calculs répétés.

   De plus, une UDF nommée count_kv est utilisée pour calculer le résultat final. Cette UDF classe et résume automatiquement les résultats statistiques dans une carte de résultats, qui est stockée au format chaîne. Cela facilite l'inférence des résultats hors ligne et en ligne ultérieure.

Le contenu précédent présente le processus de calcul des fonctions d'agrégation simples, en utilisant count et sum comme exemples. Bien que ce processus implique des concepts tels que les tables temporaires intermédiaires quotidiennes et les UDF, le flux principal est identique à celui d'une transformation de données conventionnelle et n'ajoute pas de complexité opérationnelle. D'autres fonctions d'agrégation simples, telles que max et min, sont similaires.

Processus de calcul de la fonction d'agrégation avg

Étant donné que la moyenne des moyennes quotidiennes conduit à des calculs inexacts, la fonction d'agrégation avg possède un processus de calcul spécifique. La méthode correcte consiste d'abord à calculer la somme totale (sum_v) et le nombre total (count_v) pour toute la période, puis à calculer la moyenne à l'aide de la formule sum_v/count_v.

Bien que cette fonction d'agrégation soit documentée séparément, ses détails de calcul complexes sont encapsulés dans transform_info. Vous pouvez utiliser cette fonction comme une fonctionnalité conventionnelle pour générer le résultat final.

1. Définissez les noms des tables d'entrée et de sortie.

   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. Définissez la fonctionnalité 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. Créez un pipeline et exécutez la transformation de fonctionnalité WindowTransform.

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

4. Générez et exécutez la transformation.

   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. Consultez les données de la table de destination.

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

Processus de calcul de fonction pour plusieurs clés de groupe

De même, WindowTransform prend en charge les calculs simultanés avec plusieurs clés de groupe. Le résultat est ensuite joint à gauche à la table d'entrée. Voici un exemple :

1. Définissez les noms des tables d'entrée et de sortie.

   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. Définissez les fonctionnalités 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. Créez un pipeline et exécutez la transformation de fonctionnalité 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. Générez et exécutez la transformation.

   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. Consultez les données de la table de destination.

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

Fonctions d'extension automatique intégrées pour prendre en charge l'extension automatique des fonctionnalités WindowTransform

Comme pour les transformations de fonctionnalités statistiques, la mise en œuvre manuelle de chaque fonctionnalité statistique WindowTransform est complexe en raison du grand nombre de fonctionnalités, qui incluent différentes tailles de fenêtre et de nombreuses combinaisons de calculs de fonctions d'agrégation. Le système fournit des fonctions d'extension automatique intégrées. Il vous suffit de spécifier les fonctionnalités d'entrée à compter, et le système génère et complète automatiquement les définitions de centaines de fonctionnalités statistiques.

1. Spécifiez les fonctionnalités d'entrée à compter.

   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. Définissez le nom de la table de sortie et créez un 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. Générez et exécutez la transformation.

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

Transformation JoinTransform

Dans le processus d'ingénierie des fonctionnalités précédent, en particulier pour AggregationTransform et WindowTransform, les entrées sont des tables de comportement, et les résultats de sortie sont également stockés dans une table de comportement. Cependant, la table finale utilisée en ligne n'est généralement pas une table de comportement, mais une table de fonctionnalités générée en joignant la table de comportement à d'autres tables, telles qu'une table utilisateur ou une table d'éléments.

Par conséquent, JoinTransform est introduit pour prendre en charge la jointure des fonctionnalités issues d'AggregationTransform et de WindowTransform avec des tables utilisateur ou d'éléments existantes.

Associer JoinTransform à WindowTransform

1. Définissez la table d'entrée WindowTransform.

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

2. Définissez les fonctionnalités 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. Créez un pipeline.

   Remarque

   Étant donné que d'autres tables doivent être jointes ultérieurement, la table de sortie n'est pas spécifiée ici.

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

4. Définissez les tables d'entrée et de sortie 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. Créez un pipeline JoinTransform et connectez-le au pipeline WindowTransform.

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

6. Générez et exécutez la transformation.

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

7. Consultez les données de la table de destination.

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

8. Consultez le processus de calcul réel.

   output_join_table_1.transform_info

Associer JoinTransform à AggregationTransform

1. Définissez la table d'entrée AggregationTransform.

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

2. Définissez les fonctionnalités 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. Créez un pipeline.

   Remarque

   Étant donné que d'autres tables doivent être jointes ultérieurement, la table de sortie n'est pas spécifiée ici.

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

4. Définissez les tables d'entrée et de sortie 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. Créez un pipeline JoinTransform et connectez-le au 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. Générez et exécutez la transformation.

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

7. Consultez les données de la table de destination.

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

8. Consultez le processus de calcul réel.

   output_join_table_2.transform_info