code: def test_xshards_spark_estimator(self):
      resource_path = os.path.join(
          os.path.split(__file__)[0], "../../resources"
      )
  
      def transform(df):
          result = {
              "x": np.stack(
                  [df["user"].to_numpy(), df["item"].to_numpy()], axis=1
              ),
              "y": df["label"].to_numpy(),
          }
          return result
  
      file_path = os.path.join(resource_path, "orca/learn/ncf2.csv")
      data_shard = read_csv(file_path)
      data_shard = data_shard.transform_shard(transform)
      model = Sequential()
      model.add(Linear(2, 2))
      model.add(LogSoftMax())
      optim_method = SGD(learningrate=0.01)
      with tempfile.TemporaryDirectory() as temp_dir_name:
          estimator = Estimator.from_bigdl(
              model=model,
              optimizer=optim_method,
              loss=ClassNLLCriterion(),
              metrics=Accuracy(),
              model_dir=temp_dir_name,
              feature_preprocessing=SeqToTensor([2]),
              label_preprocessing=SeqToTensor([1]),
          )
          estimator.set_constant_gradient_clipping(0.1, 1.2)
          r1 = estimator.predict(data=data_shard)
          r_c = r1.collect()
          estimator.set_tensorboard(
              log_dir=temp_dir_name, app_name="test"
          )
          estimator.fit(
              data=data_shard,
              epochs=5,
              batch_size=8,
              validation_data=data_shard,
              checkpoint_trigger=EveryEpoch(),
          )
          summary = estimator.get_train_summary(tag="Loss")
          temp_path = os.path.join(temp_dir_name, "save_model")
          estimator.save(temp_path)
          with self.assertRaises(Exception) as context:
              Estimator.from_bigdl(
                  model=model,
                  optimizer=optim_method,
                  loss=ClassNLLCriterion(),
                  metrics=["accuracy"],
                  model_dir=temp_dir_name,
                  feature_preprocessing=SeqToTensor([2]),
                  label_preprocessing=SeqToTensor([1]),
              )
          self.assertTrue(
              "Only orca metrics and customized functions are supported, but get str"
              in str(context.exception)
          )
          eval_result = estimator.evaluate(
              data=data_shard, batch_size=8
          )
          assert isinstance(eval_result, dict)
          result = estimator.predict(data=data_shard)
          assert type(result).__name__ == "SparkXShards"
          result_c = result.collect()
          df = self.get_estimator_df2()
          r0 = estimator.predict(df)
          r0_c = r0.collect()
          assert type(r0).__name__ == "DataFrame"
          for idx in range(len(r0_c)):
              assert (
                  abs(
                      r0_c[idx]["prediction"][0]
                      - result_c[0]["prediction"][idx][0]
                  )
                  <= 1e-06
              )
              assert (
                  abs(
                      r0_c[idx]["prediction"][1]
                      - result_c[0]["prediction"][idx][1]
                  )
                  <= 1e-06
              )
          estimator.fit(
              data=df,
              epochs=6,
              batch_size=8,
              validation_data=df,
              validation_trigger=EveryEpoch(),
          )
          summary = estimator.get_train_summary("Loss")
  
          # test load from checkpoint
          est2 = Estimator.from_bigdl(
              model=Sequential(),
              optimizer=None,
              loss=None,
              metrics=[Accuracy()],
              model_dir=None,
          )
          est2.load(
              temp_dir_name,
              loss=ClassNLLCriterion(),
              is_checkpoint=True,
          )
          r2 = est2.predict(data=data_shard)
          r2_c = r2.collect()
          assert (
              result_c[0]["prediction"] == r2_c[0]["prediction"]
          ).all()
          # resume training
          est2.fit(
              data=data_shard,
              epochs=10,
              batch_size=8,
              validation_data=data_shard,
              checkpoint_trigger=EveryEpoch(),
          )
          est2.evaluate(data=data_shard, batch_size=8)
          # test load from saved model
          est3 = Estimator.from_bigdl(
              model=Sequential(),
              optimizer=None,
              loss=None,
              model_dir=None,
          )
          est3.load(
              temp_path,
              optimizer=optim_method,
              loss=ClassNLLCriterion(),
          )
          r3 = est3.predict(data=data_shard)
          r3_c = r3.collect()
          assert (r3_c[0]["prediction"] == r2_c[0]["prediction"]).all()
  
imports: import shutil
  from unittest import TestCase
  from bigdl.dllib.nn.criterion import *
  from bigdl.dllib.nn.layer import *
  from pyspark.sql.types import *
  from pyspark.sql.functions import col, udf
  from bigdl.dllib.nncontext import *
  from bigdl.dllib.feature.common import *
  from bigdl.orca.learn.bigdl import Estimator
  from bigdl.dllib.optim.optimizer import Adam, SGD, ValidationSummary
  from bigdl.dllib.keras import layers
  from bigdl.dllib.keras.models import Model
  from bigdl.orca.data import SparkXShards
  from bigdl.orca.learn.metrics import Accuracy, MAE
  from bigdl.orca.learn.trigger import EveryEpoch, SeveralIteration
  from bigdl.orca.data.pandas import read_csv
label_desc: Use image node_id 9 for calling the TestEstimatorForKeras obj's underlying member method code with example usage: obj.test_xshards_spark_estimator() and returns: result