code: def prepare_features(train_tbl, test_tbl, reindex_tbls):
      def add_ratio_features(tbl):
          cal_ratio = lambda x: x[1] / x[0] if x[0] > 0 else 0.0
          tbl = tbl.apply(
              [
                  "engaged_with_user_follower_count",
                  "engaged_with_user_following_count",
              ],
              "engaged_with_user_follower_following_ratio",
              cal_ratio,
              "float",
          ).apply(
              [
                  "enaging_user_follower_count",
                  "enaging_user_following_count",
              ],
              "enaging_user_follower_following_ratio",
              cal_ratio,
              "float",
          )
          return tbl
  
      def organize_cols(tbl):
          tbl = tbl.select(
              array(
                  "enaging_user_follower_count",
                  "enaging_user_following_count",
                  "enaging_user_follower_following_ratio",
              ).alias("user_numeric"),
              array(
                  "len_hashtags",
                  "len_domains",
                  "len_links",
                  "engaged_with_user_follower_count",
                  "engaged_with_user_following_count",
                  "engaged_with_user_follower_following_ratio",
              ).alias("item_numeric"),
              *cat_cols,
              *embed_cols,
              *id_cols,
              "label",
          )
          return tbl
  
      embed_in_dims = {}
      if reindex_tbls:
          print("reindexing embedding cols")
          train_tbl = train_tbl.reindex(embed_cols, reindex_tbls)
          test_tbl = test_tbl.reindex(embed_cols, reindex_tbls)
          for (
              i,
              c,
          ) in enumerate(embed_cols):
              embed_in_dims[c] = max(
                  reindex_tbls[i]
                  .df.agg({c + "_new": "max"})
                  .collect()[0]
              )
  
      with tempfile.TemporaryDirectory() as local_path:
          get_remote_file_to_local(
              os.path.join(args.data_dir, "meta/categorical_sizes.pkl"),
              local_path,
          )
          with open(
              os.path.join(local_path, "categorical_sizes.pkl"), "rb"
          ) as f:
              cat_sizes_dict = SafePickle.load(f)
              for col in id_cols:
                  if col not in embed_in_dims:
                      embed_in_dims[col] = cat_sizes_dict[col]
  
      print("add ratio features")
      train_tbl = add_ratio_features(train_tbl)
      test_tbl = add_ratio_features(test_tbl)
  
      print("scale numerical features")
      train_tbl, min_max_dict = train_tbl.min_max_scale(
          numeric_cols + ratio_cols
      )
      test_tbl = test_tbl.transform_min_max_scale(
          numeric_cols + ratio_cols, min_max_dict
      )
  
      user_col_info = ColumnInfoTower(
          indicator_cols=["enaging_user_is_verified"],
          indicator_dims=[2],
          embed_cols=["enaging_user_id"],
          embed_in_dims=[embed_in_dims["enaging_user_id"]],
          embed_out_dims=[16],
          numerical_cols=["user_numeric"],
          numerical_dims=[3],
          name="user",
      )
      item_col_info = ColumnInfoTower(
          indicator_cols=[
              "engaged_with_user_is_verified",
              "present_media",
              "tweet_type",
              "language",
          ],
          indicator_dims=[2, 13, 3, 67],  # max + 1
          embed_cols=["tweet_id", "engaged_with_user_id"] + embed_cols,
          embed_in_dims=[
              embed_in_dims["tweet_id"],
              embed_in_dims["engaged_with_user_id"],
              embed_in_dims["hashtags"],
              embed_in_dims["present_links"],
              embed_in_dims["present_domains"],
          ],
          embed_out_dims=[16, 16, 16, 16, 16],
          numerical_cols=["item_numeric"],
          numerical_dims=[6],
          name="item",
      )
  
      print(
          "organize columns and specify user_col_info and item_col_info"
      )
      train_tbl = organize_cols(train_tbl)
      test_tbl = organize_cols(test_tbl)
  
      return train_tbl, test_tbl, user_col_info, item_col_info
  
file_path: BigDL/python/friesian/example/two_tower/train_2tower.py
imports: from bigdl.friesian.utils import SafePickle
  import argparse
  import math
  from model import *
  from pyspark.sql.functions import array
  from bigdl.orca import init_orca_context, stop_orca_context
  from bigdl.orca.data.file import get_remote_file_to_local
  from bigdl.friesian.feature import FeatureTable
  from bigdl.orca.learn.tf2.estimator import Estimator
  from bigdl.dllib.utils.log4Error import *
label_desc: Use image node_id 2 for calling a global function with example usage: prepare_features(train_tbl, test_tbl, reindex_tbls) and returns: train_tbl,  test_tbl,  user_col_info,  item_col_info, tbl, tbl