Platform For AI:Feature platform and feature engineering

Definição e execução de uma transformação de feature estatística individual

1. Defina os nomes das tabelas de entrada e saída.

   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. Defina uma feature estatística.

   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. Crie um pipeline e execute a transformação de feature estatística.

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

4. Gere e execute a transformação.

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

5. Exiba o conteúdo de transform_info.

   transform_info_agg = output_agg_table.transform_info
   transform_info_agg

6. Visualize os resultados.

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

JOIN automático para transformações de features estatísticas com janelas diferentes

1. Defina o nome da tabela de saída.

   output_agg_table_name_2 = "rec_sln_demo_behavior_test_agg_v2"

2. Defina features estatísticas para múltiplas janelas diferentes. Os tamanhos de janela definidos são 1, 3, 7, 15 e 30 dias.

   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. Crie um 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. Gere e execute o pipeline.

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

5. Consulte os resultados da transformação.

   transform_info_agg_2 = output_agg_table_2.transform_info
   transform_info_agg_2

6. Visualize os resultados da execução da tabela.

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

Mesclagem automática e derivação de tipo para múltiplos processos de transformação de features estatísticas

Para otimizar os cálculos, features com o mesmo tamanho de janela são automaticamente mescladas e calculadas no mesmo bloco de janela de grupo. O processo de cálculo envolve alterações de tipo. Por exemplo, avg transforma o tipo bigint para o tipo double. Como é difícil lembrar os tipos de todas as features de entrada, o processo de transformação de features estatísticas suporta derivação automática de tipo. O tipo da feature resultante é derivado automaticamente durante a definição da feature.

1. Defina o nome da tabela de saída.

   output_agg_table_name_3 = "rec_sln_demo_behavior_test_agg_v3"

2. Defina mais features de diferentes tipos.

   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. Crie um 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. Gere e execute o pipeline.

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

5. Consulte os resultados da transformação.

   transform_info_agg_3 = output_agg_table_3.transform_info
   transform_info_agg_3

6. Visualize os resultados da execução da tabela.

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

Funções de extensão automática integradas para transformações de features estatísticas

Implementar manualmente cada feature estatística é complexo devido ao grande número de features, que inclui diferentes tamanhos de janela e inúmeras combinações de funções de agregação. O sistema oferece funções de extensão automática integradas. Basta especificar as features de entrada a serem contabilizadas, e o sistema gera automaticamente as definições para centenas de features estatísticas.

1. Especifique as features de entrada a serem contabilizadas.

   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. Defina o nome da tabela de saída e crie um 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. Gere e execute o pipeline.

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

4. Consulte os resultados da transformação.

   transform_info_agg_4 = output_agg_table_4.transform_info
   transform_info_agg_4

5. Visualize os resultados da execução da tabela.

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

Suporte a transformação simultânea de group keys diferentes

As seções anteriores descrevem o método de processamento quando todas as group keys são iguais. O sistema também suporta operações de transformação em group keys diferentes. Veja o exemplo a seguir:

1. Defina o nome da tabela de saída.

   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. Defina features para group keys diferentes.

   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. Crie um pipeline.

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

4. Gere e execute o pipeline.

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

5. Consulte os resultados da transformação.

   transform_info_agg_5 = output_agg_table_5.transform_info
   transform_info_agg_5

Processo de cálculo de funções de agregação simples

Funções de agregação simples incluem count, sum, max e min. O processo de cálculo dessas funções é relativamente direto. Após uma sumarização em nível diário, uma sumarização adicional ao longo de múltiplos dias é realizada para obter o resultado final. Esta seção também apresenta tabelas intermediárias diárias e User-Defined Functions (UDFs) (MaxCompute UDF overview) para obter o resultado final do cálculo. No processo real de cálculo, a execução da engenharia de features funciona da mesma forma que a transformação convencional descrita anteriormente. O SDK Python do FeatureStore gerencia automaticamente a criação de tabelas intermediárias diárias, a geração de UDFs, o upload de recursos e o registro automático de funções.

1. Defina os nomes das tabelas de entrada e saída.

   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. Defina as features 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. Crie um pipeline e execute a transformação de features 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. Gere e execute a transformação.

   Este processo de geração cria uma tabela intermediária temporária diária. Neste ponto, DROP TABLE exclua apenas o resultado final, sem excluir a tabela temporária intermediária.

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

   Além disso, como estatísticas de múltiplos dias estão envolvidas (por exemplo, o exemplo anterior contabiliza dados de 7 dias), a tabela temporária intermediária normalmente calcula dados apenas para a partição mais recente. Por esse motivo, o sistema fornece o parâmetro backfill_partitions. Na primeira execução, defina este parâmetro como True para que o sistema preencha automaticamente os dados dos dias dependentes. Por exemplo, se a contagem envolve 7 dias de dados, o sistema completa automaticamente os 7 dias. Nas execuções rotineiras subsequentes, defina o parâmetro como False para completar apenas os dados da partição do dia mais recente.

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

   • Quando o parâmetro backfill_partitions é definido como True, o sistema completa automaticamente os dados dos dias dependentes da tabela intermediária temporária. Faça isso na primeira execução rotineira.

   • Se o número de dias a serem contabilizados for grande, a execução do código acima pode demorar bastante.

5. Visualize os dados da tabela de destino.

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

6. Consulte o processo real de cálculo.

   window_pipeline_1.transform_info

   Como é possível observar no processo de cálculo, o sistema gera uma tabela intermediária temporária diária chamada rec_sln_demo_behavior_table_preprocess_wide_v1_tmp_daily. Essa tabela resume os resultados diários e os armazena em uma partição fixa, evitando cálculos repetidos.

   Além disso, uma UDF chamada count_kv calcula o resultado final. Essa UDF classifica e resume automaticamente os resultados estatísticos em um mapa de resultado, armazenado em formato string. Isso facilita a inferência de resultados subsequentes, tanto offline quanto online.

O conteúdo acima apresenta o processo de cálculo para funções de agregação simples, usando count e sum como exemplos. Embora esse processo envolva conceitos como tabelas intermediárias temporárias diárias e UDFs, o fluxo principal é o mesmo de uma transformação convencional de dados e não adiciona complexidade operacional. Outras funções de agregação simples, como max e min, seguem o mesmo princípio.

Processo de cálculo da função de agregação avg

Como calcular a média dos resultados de médias diárias leva a cálculos imprecisos, a função de agregação avg possui um processo de cálculo específico. O método correto é calcular primeiro a soma total (sum_v) e a contagem total (count_v) para todo o período e então obter a média pela fórmula sum_v/count_v.

Embora essa função de agregação esteja documentada separadamente, seus detalhes complexos de cálculo estão encapsulados em transform_info. É possível usar essa função como uma feature convencional para gerar o resultado final.

1. Defina os nomes das tabelas de entrada e saída.

   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. Defina a feature 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. Crie um pipeline e execute a transformação de features WindowTransform.

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

4. Gere e execute a transformação.

   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. Visualize os dados da tabela de destino.

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

Processo de cálculo de funções para múltiplas group keys

Da mesma forma, o WindowTransform suporta cálculos simultâneos com múltiplas group keys. O resultado é então associado à tabela de entrada por meio de left join. Veja o exemplo a seguir:

1. Defina os nomes das tabelas de entrada e saída.

   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. Defina as features 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. Crie um pipeline e execute a transformação de features 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. Gere e execute a transformação.

   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. Visualize os dados da tabela de destino.

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

Funções de extensão automática integradas para features WindowTransform

De forma semelhante às transformações de features estatísticas, implementar manualmente cada feature estatística WindowTransform é complexo devido ao grande número de features, que inclui diferentes tamanhos de janela e inúmeras combinações de cálculos de funções de agregação. O sistema oferece funções de extensão automática integradas. Basta especificar as features de entrada a serem contabilizadas, e o sistema gera e completa automaticamente as definições para centenas de features estatísticas.

1. Especifique as features de entrada a serem contabilizadas.

   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. Defina o nome da tabela de saída e crie um 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. Gere e execute a transformação.

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

Transformação JoinTransform

No processo de engenharia de features descrito anteriormente, especialmente para AggregationTransform e WindowTransform, as entradas são tabelas de comportamento e os resultados de saída também são armazenados em uma tabela de comportamento. No entanto, a tabela final para uso online normalmente não é uma tabela de comportamento, mas sim uma tabela de features gerada pelo join da tabela de comportamento com outras tabelas, como a tabela de usuários ou a tabela de itens.

Por esse motivo, o JoinTransform foi introduzido para permitir o join das features de AggregationTransform e WindowTransform com tabelas existentes de usuários ou itens.

Associar JoinTransform com WindowTransform

1. Defina a tabela de entrada do WindowTransform.

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

2. Defina as features 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. Crie um pipeline.

   Nota

   Como outras tabelas precisam ser associadas posteriormente, a tabela de saída não é especificada aqui.

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

4. Defina as tabelas de entrada e saída do 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. Crie um pipeline JoinTransform e conecte-o ao pipeline WindowTransform.

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

6. Gere e execute a transformação.

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

7. Visualize os dados da tabela de destino.

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

8. Consulte o processo real de cálculo.

   output_join_table_1.transform_info

Associar JoinTransform com AggregationTransform

1. Defina a tabela de entrada do AggregationTransform.

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

2. Defina as features 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. Crie um pipeline.

   Nota

   Como outras tabelas precisam ser associadas posteriormente, a tabela de saída não é especificada aqui.

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

4. Defina as tabelas de entrada e saída do 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. Crie um pipeline JoinTransform e conecte-o ao 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. Gere e execute a transformação.

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

7. Visualize os dados da tabela de destino.

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

8. Consulte o processo real de cálculo.

   output_join_table_2.transform_info