docstr: Predict input data
  
  :param batch_size: Int. Set batch Size, default is 4.
  :param data: data to be predicted. XShards, Spark DataFrame, numpy array and list of numpy
         arrays are supported. If data is XShards, each partition is a dictionary of  {'x':
         feature}, where feature(label) is a numpy array or a list of numpy arrays.
  :param feature_cols: Feature column name(s) of data. Only used when data is a Spark
         DataFrame. Default: None.
  :return: predicted result.
           If the input data is XShards, the predict result is a XShards, each partition
           of the XShards is a dictionary of {'prediction': result}, where the result is a
           numpy array or a list of numpy arrays.
           If the input data is numpy arrays or list of numpy arrays, the predict result is
           a numpy array or a list of numpy arrays.
docstr_tok: ["Predict","input","data",":","param","batch_size",":","Int",".","Set","batch","Size",",","default","is","4.",":","param","data",":","data","to","be","predicted",".","XShards",",","Spark","DataFrame",",","numpy","array","and","list","of","numpy","arrays","are","supported",".","If","data","is","XShards",",","each","partition","is","a","dictionary","of","{","'","x","'",":","feature","}",",","where","feature","(","label",")","is","a","numpy","array","or","a","list","of","numpy","arrays",".",":","param","feature_cols",":","Feature","column","name","(","s",")","of","data",".","Only","used","when","data","is","a","Spark","DataFrame",".","Default",":","None",".",":","return",":","predicted","result",".","If","the","input","data","is","XShards",",","the","predict","result","is","a","XShards",",","each","partition","of","the","XShards","is","a","dictionary","of","{","'prediction","'",":","result","}",",","where","the","result","is","a","numpy","array","or","a","list","of","numpy","arrays",".","If","the","input","data","is","numpy","arrays","or","list","of","numpy","arrays",",","the","predict","result","is","a","numpy","array","or","a","list","of","numpy","arrays","."]
code: def predict(self, data, feature_cols=None, batch_size=4):
      """
      Predict input data
  
      :param batch_size: Int. Set batch Size, default is 4.
      :param data: data to be predicted. XShards, Spark DataFrame, numpy array and list of numpy
             arrays are supported. If data is XShards, each partition is a dictionary of  {'x':
             feature}, where feature(label) is a numpy array or a list of numpy arrays.
      :param feature_cols: Feature column name(s) of data. Only used when data is a Spark
             DataFrame. Default: None.
      :return: predicted result.
               If the input data is XShards, the predict result is a XShards, each partition
               of the XShards is a dictionary of {'prediction': result}, where the result is a
               numpy array or a list of numpy arrays.
               If the input data is numpy arrays or list of numpy arrays, the predict result is
               a numpy array or a list of numpy arrays.
      """
      sc = init_nncontext()
      model_bytes_broadcast = sc.broadcast(self.model_bytes)
      weight_bytes_broadcast = sc.broadcast(self.weight_bytes)
  
      def partition_inference(partition):
          model_bytes = model_bytes_broadcast.value
          weight_bytes = weight_bytes_broadcast.value
          partition = list(partition)
          data_num = len(partition)
          ie = IECore()
          config = {"CPU_THREADS_NUM": str(self.core_num)}
          ie.set_config(config, "CPU")
          net = ie.read_network(
              model=model_bytes,
              weights=weight_bytes,
              init_from_buffer=True,
          )
          net.batch_size = batch_size
          local_model = ie.load_network(
              network=net, device_name="CPU", num_requests=data_num
          )
          inputs = list(iter(local_model.requests[0].input_blobs))
          outputs = list(iter(local_model.requests[0].output_blobs))
          assert (
              len(outputs) != 0
          ), "The number of model outputs should not be 0."
  
          def add_elem(d):
              d_len = len(d)
              if d_len < batch_size:
                  rep_time = [1] * (d_len - 1)
                  rep_time.append(batch_size - d_len + 1)
                  return np.repeat(d, rep_time, axis=0), d_len
              else:
                  return d, d_len
  
          results = []
          for idx, batch_data in enumerate(partition):
              infer_request = local_model.requests[idx]
              input_dict = dict()
              elem_num = 0
              if isinstance(batch_data, list):
                  for i, input in enumerate(inputs):
                      input_dict[input], elem_num = add_elem(
                          batch_data[i]
                      )
              else:
                  input_dict[inputs[0]], elem_num = add_elem(batch_data)
              infer_request.infer(input_dict)
              if len(outputs) == 1:
                  results.append(
                      infer_request.output_blobs[outputs[0]].buffer[
                          :elem_num
                      ]
                  )
              else:
                  results.append(
                      list(
                          map(
                              lambda output: infer_request.output_blobs[
                                  output
                              ].buffer[:elem_num],
                              outputs,
                          )
                      )
                  )
  
          return results
  
      def predict_transform(dict_data, batch_size):
          assert isinstance(
              dict_data, dict
          ), "each shard should be an dict"
          assert "x" in dict_data, "key x should in each shard"
          feature_data = dict_data["x"]
          if isinstance(feature_data, np.ndarray):
              assert feature_data.shape[0] <= batch_size, (
                  "The batch size of input data (the second dim) should be less than the model "
                  "batch size, otherwise some inputs will be ignored."
              )
          elif isinstance(feature_data, list):
              for elem in feature_data:
                  assert isinstance(elem, np.ndarray), (
                      "Each element in the x list should be "
                      "a ndarray, but get " + elem.__class__.__name__
                  )
                  assert elem.shape[0] <= batch_size, (
                      "The batch size of each input data (the "
                      "second dim) should be less than the "
                      "model batch size, otherwise some inputs "
                      "will be ignored."
                  )
          else:
              raise ValueError(
                  "x in each shard should be a ndarray or a list of ndarray."
              )
          return feature_data
  
      if isinstance(data, DataFrame):
          from zoo.orca.learn.utils import (
              dataframe_to_xshards,
              convert_predict_rdd_to_dataframe,
          )
  
          xshards, _ = dataframe_to_xshards(
              data,
              validation_data=None,
              feature_cols=feature_cols,
              label_cols=None,
              mode="predict",
          )
          transformed_data = xshards.transform_shard(
              predict_transform, batch_size
          )
          result_rdd = transformed_data.rdd.mapPartitions(
              lambda iter: partition_inference(iter)
          )
          return convert_predict_rdd_to_dataframe(
              data, result_rdd.flatMap(lambda data: data)
          )
      elif isinstance(data, SparkXShards):
          transformed_data = data.transform_shard(
              predict_transform, batch_size
          )
          result_rdd = transformed_data.rdd.mapPartitions(
              lambda iter: partition_inference(iter)
          )
  
          def update_result_shard(data):
              shard, y = data
              shard["prediction"] = y
              return shard
  
          return SparkXShards(
              data.rdd.zip(result_rdd).map(update_result_shard)
          )
      elif isinstance(data, (np.ndarray, list)):
          if isinstance(data, np.ndarray):
              split_num = math.ceil(len(data) / batch_size)
              arrays = np.array_split(data, split_num)
              num_slices = min(split_num, self.node_num)
              data_rdd = sc.parallelize(arrays, numSlices=num_slices)
          elif isinstance(data, list):
              flattened = nest.flatten(data)
              data_length = len(flattened[0])
              data_to_be_rdd = []
              split_num = math.ceil(flattened[0].shape[0] / batch_size)
              num_slices = min(split_num, self.node_num)
              for i in range(split_num):
                  data_to_be_rdd.append([])
              for x in flattened:
                  assert isinstance(x, np.ndarray), (
                      "the data in the data list should be "
                      "ndarrays, but get " + x.__class__.__name__
                  )
                  assert len(x) == data_length, (
                      "the ndarrays in data must all have the same size in first dimension"
                      ", got first ndarray of size {} and another {}".format(
                          data_length, len(x)
                      )
                  )
                  x_parts = np.array_split(x, split_num)
                  for idx, x_part in enumerate(x_parts):
                      data_to_be_rdd[idx].append(x_part)
  
              data_to_be_rdd = [
                  nest.pack_sequence_as(data, shard)
                  for shard in data_to_be_rdd
              ]
              data_rdd = sc.parallelize(
                  data_to_be_rdd, numSlices=num_slices
              )
  
          print("Partition number: ", data_rdd.getNumPartitions())
          result_rdd = data_rdd.mapPartitions(
              lambda iter: partition_inference(iter)
          )
          result_arr_list = result_rdd.collect()
          result_arr = None
          if isinstance(result_arr_list[0], list):
              result_arr = [
                  np.concatenate(
                      [r[i] for r in result_arr_list], axis=0
                  )
                  for i in range(len(result_arr_list[0]))
              ]
          elif isinstance(result_arr_list[0], np.ndarray):
              result_arr = np.concatenate(result_arr_list, axis=0)
          return result_arr
      else:
          raise ValueError(
              "Only XShards, Spark DataFrame, a numpy array and a list of numpy arr"
              "ays are supported as input data, but get "
              + data.__class__.__name__
          )
  
imports: import math
  import os.path
  from pyspark.sql import DataFrame
  from zoo.orca.data import SparkXShards
  from zoo.orca.learn.spark_estimator import Estimator
  from zoo import get_node_and_core_number
  from zoo.util import nest
  from zoo.common.nncontext import init_nncontext
  from openvino.inference_engine import IECore
  import numpy
label_desc: Use image node_id 3 for calling the OpenvinoEstimator obj's underlying member method code with example usage: obj.predict(data, feature_cols, batch_size) and returns: results, feature_data, convert_predict_rdd_to_dataframe, SparkXShards, np,  d_len, d,  d_len, shard, result_arr