do we support scala & java code write tensorflow model with tenorflow-core-api ? about tensorflowonspark HOT 3 OPEN

  override def transform(dataset: Dataset[_]): DataFrame = {
    val spark = dataset.sparkSession

    val inputColumns = this.getInputMapping.keys.toSeq
    val inputTensorNames = this.getInputMapping.values
    val outputTensorNames = this.getOutputMapping.keys.toSeq

    val inputDF = dataset.select(inputColumns.head, inputColumns.tail: _*)
    val inputSchema = inputDF.schema
    val outputSchema = transformSchema(inputSchema)

    val outputRDD = inputDF.rdd.mapPartitions { iter: Iterator[Row] =>
      if (TFModel.model == null || TFModel.modelDir != this.getModel) {
        // load model into a per-executor singleton reference, if needed.
        TFModel.modelDir = this.getModel
        TFModel.model = SavedModelBundle.load(this.getModel, this.getTag)
        TFModel.graph = TFModel.model.graph
        TFModel.sess = TFModel.model.session
      }

      iter.grouped(this.getBatchSize).flatMap { batch =>
        // get input batch of Rows and convert to list of input Tensors
        val inputTensors = batch2tensors(batch, inputSchema)

        var runner = TFModel.sess.runner()

        // feed input tensors
        for ((name, tensor) <- inputTensors) {
          runner = runner.feed(this.getInputMapping(name), tensor)
        }
        // fetch output tensors
        for (name <- outputTensorNames) {
          runner = runner.fetch(name)
        }

        // run the graph
        val outputTensors = runner.run()

        assert(outputTensors.map(_.shape).map(s => if (s.isEmpty) 0L else s.apply(0)).distinct.size == 1,
          "Cardinality of output tensors must match")

        // convert the list of output Tensors to a batch of output Rows
        tensors2batch(outputTensors)
      }
    }

    spark.createDataFrame(outputRDD, outputSchema)
  }