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_num"),
              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_num"),
              *cat_cols,
              *embed_cols,
              "label",
          )
          return tbl
  
      print("reindexing embedding cols")
      train_tbl = train_tbl.reindex(embed_cols, reindex_tbls)
      test_tbl = test_tbl.reindex(embed_cols, reindex_tbls)
      embed_in_dims = {}
      for (
          i,
          c,
      ) in enumerate(embed_cols):
          embed_in_dims[c] = max(
              reindex_tbls[i].df.agg({c + "_new": "max"}).collect()[0]
          )
  
      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_dic = train_tbl.min_max_scale(
          num_cols + ratio_cols
      )
      test_tbl = test_tbl.transform_min_max_scale(
          num_cols + ratio_cols, min_max_dic
      )
  
      with open(
          os.path.join(args.model_dir, "stats/min_max.pkl"), "wb"
      ) as f:
          pickle.dump(min_max_dic, f)
  
      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_num"],
          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=[
              "engaged_with_user_id",
              "hashtags",
              "present_links",
              "present_domains",
          ],
          embed_in_dims=[
              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],
          numerical_cols=["item_num"],
          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: analytics-zoo/pyzoo/zoo/examples/friesian/two_tower/train_2tower.py
imports: import pickle
  import argparse
  from pyspark.sql.functions import array
  from zoo.orca import init_orca_context
  from zoo.orca.learn.tf2.estimator import Estimator
  from zoo.friesian.feature import FeatureTable
  from model 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