6-7 Вызов модели TensorFlow с использованием spark-scala

import tensorflow as tf
from tensorflow.keras import models,layers,optimizers

## Количество образцов
n = 800

## Генерация тестового набора данных
X = tf.random.uniform([n,2],minval=-10,maxval=10) 
w0 = tf.constant([[2.0],[-1.0]])
b0 = tf.constant(3.0)

Y = X@w0 + b0 + tf.random.normal([n,1],mean = 0.0,stddev= 2.0)  # @ — умножение матриц; добавление гауссовского шума

## Моделирование
tf.keras.backend.clear_session()
inputs = layers.Input(shape = (2,),name ="inputs") # Установить имя ввода как "входы"
outputs = layers.Dense(1, name = "outputs")(inputs) # Установить выходное имя как "выходы"
linear = models.Model(inputs = inputs,outputs = outputs)
linear.summary()

## Обучение с методом подгонки
linear.compile(optimizer="rmsprop",loss="mse",metrics=["mae"])
linear.fit(X,Y,batch_size = 8,epochs = 100)  

tf.print("w = ",linear.layers[1].kernel)
tf.print("b = ",linear.layers[1].bias)

## Сохранить модель в формате pb
export_path = "../data/linear_model/"
version = "1"       # Версия может использоваться для управления дальнейшими обновлениями
linear.save(export_path+version, save_format="tf") 

!ls {export_path+version}

# Проверить информацию о файле модели
!saved_model_cli show --dir {export_path+str(version)} --all

<!-- https://mvnrepository.com/artifact/org.tensorflow/tensorflow -->
<dependency>
    <groupId>org.tensorflow</groupId>
    <artifactId>tensorflow</artifactId>
    <version>1.15.0</version>
</dependency>

Array(Array(3.019596), Array(3.9878292))

![](../data/TfDriver.png)

import org.apache.spark.sql.SparkSession
import scala.collection.mutable.WrappedArray
import org.{tensorflow=>tf}

val spark = SparkSession
    .builder()
    .appName("TfRDD")
    .enableHiveSupport()
    .getOrCreate()

val sc = spark.sparkContext

// Загрузка модели на стороне драйвера
val bundle = tf.SavedModelBundle 
   .load("/Users/liangyun/CodeFiles/master_tensorflow2_in_20_hours/data/linear_model/1","serve")

// Широковещательная рассылка модели всем исполнителям
val broads = sc.broadcast(bundle)

// Создание набора данных
val rdd_data = sc.makeRDD(List(Array(1.0f,2.0f),Array(3.0f,5.0f),Array(6.0f,7.0f),Array(8.0f,3.0f)))

// Прогнозирование в пакетном режиме с использованием модели через mapPartitions
val rdd_result = rdd_data.mapPartitions(iter => {
    
    val arr = iter.toArray
    val model = broads.value
    val sess = model.session()
    val x = tf.Tensor.create(arr)
    val y =  sess.runner().feed("serving_default_inputs:0", x)
             .fetch("StatefulPartitionedCall:0").run().get(0)

    // Копирование прогноза в массив типа Float с той же формой
    val result = Array.ofDim[Float](y.shape()(0).toInt,y.shape()(1).toInt)
    y.copyTo(result)
    result.iterator
    
})

rdd_result.take(5)
bundle.close

<!-- #endregion -->

Array(Array(3.019596), Array(3.9264367), Array(7.8607616), Array(15.974984))

![](../data/TfRDD.png)

import org.apache.spark.sql.SparkSession
import scala.collection.mutable.WrappedArray
import org.{tensorflow=>tf}

object TfDataFrame extends Serializable{
    
    def main(args:Array[String]):Unit = {
        
        val spark = SparkSession
        .builder()
        .appName("TfDataFrame")
        .enableHiveSupport()
        .getOrCreate()
        val sc = spark.sparkContext
        
        import spark.implicits._

        val bundle = tf.SavedModelBundle 
           .load("/Users/liangyun/CodeFiles/master_tensorflow2_in_20_hours/data/linear_model/1","serve")

        val broads = sc.broadcast(bundle)
        
        // Создаем функцию прогнозирования и регистрируем ее как udf в sparkSQL
        val tfpredict = (features:WrappedArray[Float])  => {
            val bund = broads.value
            val sess = bund.session()
            val x = tf.Tensor.create(Array(features.toArray))
            val y =  sess.runner().feed("serving_default_inputs:0", x)
                   .fetch("StatefulPartitionedCall:0").run().get(0)
``` ```
val result = Array.ofDim[Float](y.shape()(0).toInt, y.shape()(1).toInt)
y.copyTo(result)
val y_pred = result(0)(0)
y_pred
}
spark.udf.register("tfpredict", tfpredict)

// Creating DataFrame dataset, and put the features into one of the columns
val dfdata = sc.parallelize(List(Array(1.0f, 2.0f), Array(3.0f, 5.0f), Array(7.0f, 8.0f))).toDF("features")
dfdata.show

// Call the sparkSQL predicting function, add a new column as y_preds
val dfresult = dfdata.selectExpr("features", "tfpredict(features) as y_preds")
dfresult.show
bundle.close
}
}

+----------+
|  features|
+----------+
|[1.0, 2.0]|
|[3.0, 5.0]|
|[7.0, 8.0]|
+----------+

+----------+---------+
|  features|  y_preds|
+----------+---------+
|[1.0, 2.0]| 3.019596|
|[3.0, 5.0]|3.9264367|
|[7.0, 8.0]| 8.828995|
+----------+---------+