如何基于 Byzer 使用深度学习快速开发一个图片分类应用

-- 加载图片数据
load image.`/trainingSet/trainingSet/**/*.jpg` as mnist;

-- 对图片做分区
run mnist as TableRepartition.`` where partitionNum="4" as newMnist;

-- 后保存到数据湖里去
save overwrite newMnist as delta.`data.raw_mnist` where mergeSchema="true";

-- 从数据湖加载数据
load delta.`data.raw_mnist` as mnist;

-- 从路径中获得分类标签
select split(image.origin,"/")[7] as label, image.data as content from mnist as newMnist;

#%python
#%input=newMnist
#%output=newMnist_model
#%cache=true
#%schema=file
#%dataMode=model
#%env=source /home/williamzhu/miniconda3/bin/activate ray-1.12.0

from functools import reduce
import os
import ray
import numpy as np
from tensorflow.keras import models,layers
from tensorflow.keras import utils as np_utils
from pyjava.api.mlsql import RayContext
from pyjava.storage import streaming_tar

## 连接 ray 集群
ray_context = RayContext.connect(globals(),"127.0.0.1:10001")

## 获取所有数据分片地址
data_servers = ray_context.data_servers()
replica_num = len(data_servers)

## 获取每个数据分片的数据
def data_partition_creater(data_server):
    temp_data = [item for item in RayContext.collect_from([data_server])]
    
    train_images = np.array([np.array(list(item["content"])) for item in temp_data])    
    train_labels = np_utils.to_categorical(np.array([item["label"] for item in temp_data]))
    train_images = train_images.reshape((len(temp_data),28*28))
    return train_images,train_labels    

## 创建模型
def create_tf_model():    
    network = models.Sequential()
    network.add(layers.Dense(512,activation="relu",input_shape=(28*28,)))
    network.add(layers.Dense(10,activation="softmax"))
    network.compile(optimizer="sgd",loss="categorical_crossentropy",metrics=["accuracy"])
    return network

## 构建训练类
@ray.remote
class Network(object):
    def __init__(self,data_server):
        self.model = create_tf_model()
        # you can also save the data to local disk if the data is 
        # not fit in memory        
        self.train_images,self.train_labels = data_partition_creater(data_server)


    def train(self):
        history = self.model.fit(self.train_images,self.train_labels,batch_size=128)
        return history.history

    def get_weights(self):
        return self.model.get_weights()

    def set_weights(self, weights):
        # Note that for simplicity this does not handle the optimizer state.
        self.model.set_weights(weights)

    def get_final_model(self):
        model_path = os.path.join("/","tmp","minist_model")
        self.model.save(model_path)
        model_binary = [item for item in streaming_tar.build_rows_from_file(model_path)]
        return model_binary
    def shutdown(self):
        ray.actor.exit_actor()

## 实例化训练类，有多少个数据分片，就会实例化多少个训练类，并且他们都是在远程引用
workers = [Network.remote(data_server) for data_server in data_servers]
ray.get([worker.train.remote() for worker in workers])

## 获取初始训练的权重
_weights = ray.get([worker.get_weights.remote() for worker in workers])

## 对不同实例计算得到的权重做加权平均，然后返回给每个实例，再进行计算
def epoch_train(weights):  
    sum_weights = reduce(lambda a,b: [(a1 + b1) for a1,b1 in zip(a,b)],weights)
    averaged_weights = [layer/replica_num for layer in sum_weights]
    ray.get([worker.set_weights.remote(averaged_weights) for worker in workers])
    ray.get([worker.train.remote() for worker in workers])
    return ray.get([worker.get_weights.remote() for worker in workers])

for epoch in range(30):
   _weights = epoch_train(_weights)

## 随机取一个模型的权重，然后输出
model_binary = ray.get(workers[0].get_final_model.remote())
[worker.shutdown.remote() for worker in workers]
ray_context.build_result(model_binary)

save overwrite newMnist_model as delta.`ai_model.mnist_model`;

load delta.`ai_model.mnist_model` as mnist_model;

!python conf "rayAddress=127.0.0.1:10001";
!python conf "schema=file";
!python env "PYTHON_ENV=source /home/williamzhu/miniconda3/bin/activate ray-1.12.0";
!python conf "dataMode=model";
!python conf "runIn=driver";

register Ray.`mnist_model` as mnist_model_predict where 
maxConcurrency="2"
and debugMode="true"
and registerCode='''

import ray
import numpy as np
from pyjava.api.mlsql import RayContext
from pyjava.udf import UDFMaster,UDFWorker,UDFBuilder,UDFBuildInFunc

ray_context = RayContext.connect(globals(), context.conf["rayAddress"])

## 这个是预测函数，函数会给你模型以及待预测向量
def predict_func(model,v):
    train_images = np.array([v])
    train_images = train_images.reshape((1,28*28))
    predictions = model.predict(train_images)
    return {"value":[[float(np.argmax(item)) for item in predictions]]}

UDFBuilder.build(ray_context,UDFBuildInFunc.init_tf,predict_func)

''' and 
predictCode='''

import ray
from pyjava.api.mlsql import RayContext
from pyjava.udf import UDFMaster,UDFWorker,UDFBuilder,UDFBuildInFunc

ray_context = RayContext.connect(globals(), context.conf["rayAddress"])
UDFBuilder.apply(ray_context)

'''
;

select mnist_model_predict(array(byteArrayToIntArray(content)))[0][0] as predicted_label,label 
from newMnist limit 10 
as output;

-- 控制这个 Form App 的访问性。在这里，只允许william访问
set FORM_VISIBILITY = "user:william";

-- 设置 Form App 的标题
set title = "图片预测" where type="defaultParam" and formType="title" ;

-- 设置 Form App 的表单元素，这里我们需要一个让用户上传图片的框
set pic="" where 
type="defaultParam" 
and formType="upload" 
and label="上传图片";

set FORM_VISIBILITY = "user:william";

set title = "图片预测" where type="defaultParam" and formType="title" ;

set pic="" where 
type="defaultParam" 
and formType="upload" 
and label="上传图片";

select decodeImage(unbase64("${pic}")) as content, "wow1.png" as fileName as imageTable;


select mnist_model_predict(array(byteArrayToIntArray(content)))[0][0]  as predicted_label 
from imageTable 
as output;