Platform For AI:Feature platform and feature engineering

Definition and running of a single statistical feature transformation

1. Define the input and output table names.

   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. Define a statistical feature.

   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="<>"), # Set a condition that is met when the value of the "event" field is not equal to "expr". You can understand the specific logic from the generated SQL statement.
                   group_by_keys="user_id", # The key corresponding to group by.
                   window_size=DayOf(1), # The window size, which is 1 day here.
               ),
           )

3. You can create a pipeline and run the statistical feature transformation.

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

4. Generate and execute the transformation.

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

5. View the content of transform_info.

   transform_info_agg = output_agg_table.transform_info
   transform_info_agg

6. View the results.

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

Automatic JOIN for statistical feature transformations with different windows

1. Define the output table name.

   output_agg_table_name_2 = "rec_sln_demo_behavior_test_agg_v2"

2. Define statistical features for multiple different windows. The defined window sizes are 1, 3, 7, 15, and 30 days.

   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. Create a 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. Generate and execute the pipeline.

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

5. View the transformation results.

   transform_info_agg_2 = output_agg_table_2.transform_info
   transform_info_agg_2

6. View the results of the table run.

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

Automatic merging and type derivation for multiple statistical feature transformation processes

To optimize calculations, multiple features with the same window size are automatically merged and calculated in the same group window block. The calculation process involves type changes. For example, `avg` transforms the `bigint` type to the `double` type. It is difficult to remember the numerous types of input features. Therefore, the statistical feature transformation process supports automatic type derivation, which means you do not need to specify the type in advance. The type of the resulting feature is automatically derived during the feature definition.

1. Define the output table name.

   output_agg_table_name_3 = "rec_sln_demo_behavior_test_agg_v3"

2. Define more features of different 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. Create a 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. Generate and execute the pipeline.

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

5. View the transformation results.

   transform_info_agg_3 = output_agg_table_3.transform_info
   transform_info_agg_3

6. View the results of the table run.

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

Built-in automatic extension functions to support automatic extension of statistical feature transformations

Manually implementing each statistical feature is complex because of the large number of features, which include different window sizes and numerous combinations of aggregate function calculations. The system provides built-in automatic extension functions. You only need to specify the input features to be counted, and the system automatically generates and completes the definitions for hundreds of statistical features.

1. Specify the input features to be counted.

   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. Define the output table name and create a 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. Generate and execute the pipeline.

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

4. View the transformation results.

   transform_info_agg_4 = output_agg_table_4.transform_info
   transform_info_agg_4

5. View the results of the table run.

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

Support for simultaneous transformation of different group keys

The previous sections describe the processing method for when all group keys are the same. The system also supports transformation operations on different group keys. The following is an example:

1. Define the output table name.

   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. Define features for different group keys.

   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. Create a pipeline.

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

4. Generate and execute the pipeline.

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

5. View the transformation results.

   transform_info_agg_5 = output_agg_table_5.transform_info
   transform_info_agg_5

Simple aggregate function calculation process

Simple aggregate functions include count, sum, max, and min. The calculation process for these aggregate functions is relatively straightforward. After a daily-level summarization, a further summarization over multiple days is performed to obtain the final result. In addition, this section also introduces daily intermediate tables and the use of User-Defined Functions (UDFs) (MaxCompute UDF overview) to obtain the final calculation result. In the actual calculation process, the process for a user to execute feature engineering is the same as the conventional feature transformation described previously. The FeatureStore Python SDK automatically handles the creation of daily intermediate tables, UDF generation, resource upload, and automatic function registration. You can implement the feature engineering process without being aware of these details.

1. Define the input and output table names.

   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. Define the WindowTransform features.

   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. Create a pipeline and run the WindowTransform feature transformation.

   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. Generate and execute the transformation.

   This generation process creates a daily temporary intermediate table. At this point, `DROP TABLE` only deletes the final result and does not delete the intermediate temporary table.

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

   In addition, because statistics for multiple days are involved (for example, the preceding example counts data for 7 days), the intermediate temporary table typically calculates data only for the latest partition. For this reason, the system provides the `backfill_partitions` parameter. When you execute the process for the first time, you can set this parameter to `True`, and the system automatically backfills the data for the dependent days. For example, if the count involves 7 days of data, the system automatically completes the 7 days of data. For subsequent routine runs, you can set the parameter to `False` to complete the data only for the latest day's partition.

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

   • When the `backfill_partitions` parameter is set to `True`, the system automatically completes the data for the dependent days of the temporary intermediate table. We recommend that you do this during the first routine run.

   • If the number of days to be counted is large, running the preceding code takes a long time.

5. View the sink table data.

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

6. View the actual calculation process.

   window_pipeline_1.transform_info

   As you can see from the calculation process, the system generates a daily intermediate temporary table named `rec_sln_demo_behavior_table_preprocess_wide_v1_tmp_daily`. This table summarizes the daily results and stores them in a fixed partition, which avoids repeated calculations.

   In addition, a UDF named `count_kv` is used to calculate the final result. This UDF automatically classifies and summarizes the statistical results into a result map, which is stored in a string format. This facilitates subsequent offline and online result inference.

The preceding content introduces the calculation process for simple aggregate functions, using `count` and `sum` as examples. Although this process involves concepts such as daily intermediate temporary tables and UDFs, the core flow is the same as a conventional data transformation operation and does not add operational complexity. Other simple aggregate functions, such as `max` and `min`, are similar.

avg aggregate function calculation process

Because averaging the results of daily averages leads to inaccurate calculations, the avg aggregate function has a specific calculation process. The correct method is to first calculate the total sum (sum_v) and total count (count_v) for the entire period, and then calculate the average using the formula sum_v/count_v.

Although this aggregate function is documented separately, its complex calculation details are encapsulated in transform_info. Therefore, you do not need to understand the underlying details. You can use this function like a conventional feature to generate the final result.

1. Define the input and output table names.

   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. You can define the WindowTransform feature.

   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. Create a pipeline and run the WindowTransform feature transformation.

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

4. Generate and execute the 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. View the sink table data.

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

Function calculation process for multiple group keys

Similarly, WindowTransform supports simultaneous calculations with multiple group keys. The result is then left-joined to the input table. The following is an example:

1. Define the input and output table names.

   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. Define the WindowTransform features.

   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. Create a pipeline and run the WindowTransform feature transformation.

   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. Generate and execute the 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. View the sink table data.

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

Built-in automatic extension functions to support automatic extension of WindowTransform features

Similar to statistical feature transformations, manually implementing each WindowTransform statistical feature is complex because of the large number of features, which include different window sizes and numerous combinations of aggregate function calculations. The system provides built-in automatic extension functions. You only need to specify the input features to be counted, and the system automatically generates and completes the definitions for hundreds of statistical features.

1. Specify the input features to be counted.

   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. Define the output table name and create a 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. Generate and execute the 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)

JoinTransform transformation

In the preceding feature engineering process, especially for AggregationTransform and WindowTransform, the inputs are behavior tables, and the output results are also stored in a behavior table. However, in most cases, the final table needed for online use is not a behavior table, but a feature table generated by joining the behavior table with other tables, such as a user table or an item table.

Therefore, `JoinTransform` is introduced to support joining the features from AggregationTransform and WindowTransform with existing user or item tables.

Associate JoinTransform with WindowTransform

1. Define the WindowTransform input table.

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

2. Define the WindowTransform features.

   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. Create a pipeline.

   Note

   Because other tables need to be joined later, the output table is not specified here.

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

4. Define the JoinTransform input and output tables.

   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. Create a JoinTransform pipeline and connect it with the WindowTransform pipeline.

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

6. Generate and execute the transformation.

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

7. View the sink table data.

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

8. View the actual calculation process.

   output_join_table_1.transform_info

Associate JoinTransform with AggregationTransform

1. Define the AggregationTransform input table.

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

2. Define the AggregationTransform features.

   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. Create a pipeline.

   Note

   Because other tables need to be joined later, the output table is not specified here.

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

4. Define the JoinTransform input and output tables.

   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. Create a JoinTransform pipeline and connect it with the AggregationTransform pipeline.

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

6. Generate and execute the transformation.

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

7. View the sink table data.

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

8. View the actual calculation process.

   output_join_table_2.transform_info