code: def test_xshards_pandas_dataframe(self):
      from bigdl.orca.data.pandas import read_csv
  
      sc = OrcaContext.get_spark_context()
  
      def model_creator(config):
          import tensorflow as tf
  
          input1 = tf.keras.layers.Input(shape=(1,))
          input2 = tf.keras.layers.Input(shape=(1,))
          concatenation = tf.concat([input1, input2], axis=-1)
          outputs = tf.keras.layers.Dense(
              units=1, activation="softmax"
          )(concatenation)
          model = tf.keras.Model(
              inputs=[input1, input2], outputs=outputs
          )
          model.compile(**compile_args(config))
          return model
  
      resource_path = os.path.join(
          os.path.split(__file__)[0], "../../resources"
      )
      file_path = os.path.join(resource_path, "orca/learn/ncf.csv")
  
      xshards = read_csv(
          file_path,
          usecols=[0, 1, 2],
          dtype={0: np.float32, 1: np.float32, 2: np.float32},
      )
  
      config = {"lr": 0.2}
  
      try:
          temp_dir = tempfile.mkdtemp()
  
          trainer = Estimator.from_keras(
              model_creator=model_creator,
              verbose=True,
              config=config,
              workers_per_node=1,
              backend="spark",
          )
  
          res = trainer.fit(
              data=xshards,
              epochs=5,
              batch_size=4,
              steps_per_epoch=25,
              feature_cols=["user", "item"],
              label_cols=["label"],
          )
  
          assert isinstance(res, dict), "fit should return a dict"
          print("start saving")
          trainer.save_weights(
              os.path.join(temp_dir, "cifar10_keras.h5")
          )
          trainer.load_weights(
              os.path.join(temp_dir, "cifar10_keras.h5")
          )
          res = trainer.evaluate(
              data=xshards,
              num_steps=25,
              batch_size=4,
              feature_cols=["user", "item"],
              label_cols=["label"],
          )
          assert isinstance(res, dict), "evaluate should return a dict"
          print("validation result: ", res)
          predictions = trainer.predict(
              data=xshards, batch_size=4, feature_cols=["user", "item"]
          )
          assert (
              predictions._get_class_name()
              == "pandas.core.frame.DataFrame"
          )
          prediction_df = predictions.collect()[0]
          import pandas as pd
  
          assert isinstance(prediction_df, pd.DataFrame)
          assert "prediction" in prediction_df
      finally:
          shutil.rmtree(temp_dir)
  
imports: import os
  import pytest
  import shutil
  import tempfile
  from unittest import TestCase
  import numpy
  import tensorflow
  from bigdl.orca.learn.tf2 import Estimator
  from bigdl.orca import OrcaContext
label_desc: Use image node_id 2 for calling the TestTF2Estimator obj's underlying member method code with example usage: obj.test_xshards_pandas_dataframe() and returns: model