<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 In this example, we will use tensorflow.keras package to create a keras image classification application using model MobileNetV2, and transfer the application to Cluster Serving step by step.
Original Keras application
We will first show an original Keras application, which download the data and preprocess it, then create the MobileNetV2 model to predict.
Step1:  In keras, input could be ndarray, or generator. We could just use model.predict(test_generator). But to simplify, here we just input the first record to model.
Step2:  Great! Now the Keras application is completed. 
Export TensorFlow SavedModel
Next, we transfer the application to Cluster Serving. The first step is to save the model to SavedModel format.
Step3:  Deploy Cluster Serving
After model prepared, we start to deploy it on Cluster Serving.
First install Cluster Serving
Step4:  We config the model path in config.yaml to following (the detail of config is at Cluster Serving Configuration)
Step5:  Start Cluster Serving
Cluster Serving requires Flink and Redis installed, and corresponded environment variables set, check Cluster Serving Installation Guide for detail.
Flink cluster should start before Cluster Serving starts, if Flink cluster is not started, call following to start a local Flink cluster.
Step6:  After configuration, start Cluster Serving by cluster-serving-start (the detail is at Cluster Serving Programming Guide)
Step7:  Prediction using Cluster Serving
Next we start Cluster Serving code at python client.
Step8:  In Cluster Serving, only NdArray is supported as input. Thus, we first transform the generator to ndarray (If you do not know how to transform your input to NdArray, you may get help at data transform guide)
Step9:  If everything works well, the result prediction would be the exactly the same NdArray object with the output of original Keras model.
<ASSISTANT_TASK:>
Python Code:
import tensorflow as tf
import os
import PIL
tf.__version__
# Obtain data from url:"https://storage.googleapis.com/mledu-datasets/cats_and_dogs_filtered.zip"
zip_file = tf.keras.utils.get_file(origin="https://storage.googleapis.com/mledu-datasets/cats_and_dogs_filtered.zip",
                                   fname="cats_and_dogs_filtered.zip", extract=True)
# Find the directory of validation set
base_dir, _ = os.path.splitext(zip_file)
test_dir = os.path.join(base_dir, 'validation')
# Set images size to 160x160x3
image_size = 160
# Rescale all images by 1./255 and apply image augmentation
test_datagen = tf.keras.preprocessing.image.ImageDataGenerator(rescale=1./255)
# Flow images using generator to the test_generator
test_generator = test_datagen.flow_from_directory(
                test_dir,
                target_size=(image_size, image_size),
                batch_size=1,
                class_mode='binary')
# Create the base model from the pre-trained model MobileNet V2
IMG_SHAPE=(160,160,3)
model = tf.keras.applications.MobileNetV2(input_shape=IMG_SHAPE,
                                               include_top=False,
                                               weights='imagenet')
Explanation: In this example, we will use tensorflow.keras package to create a keras image classification application using model MobileNetV2, and transfer the application to Cluster Serving step by step.
Original Keras application
We will first show an original Keras application, which download the data and preprocess it, then create the MobileNetV2 model to predict.
End of explanation
prediction=model.predict(test_generator.next()[0])
print(prediction)
Explanation: In keras, input could be ndarray, or generator. We could just use model.predict(test_generator). But to simplify, here we just input the first record to model.
End of explanation
# Save trained model to ./transfer_learning_mobilenetv2
model.save('/tmp/transfer_learning_mobilenetv2')
! ls /tmp/transfer_learning_mobilenetv2
Explanation: Great! Now the Keras application is completed. 
Export TensorFlow SavedModel
Next, we transfer the application to Cluster Serving. The first step is to save the model to SavedModel format.
End of explanation
! pip install bigdl-serving
# we go to a new directory and initialize the environment
! mkdir cluster-serving
os.chdir('cluster-serving')
! cluster-serving-init
! tail wget-log.2
# if you encounter slow download issue like above, you can just use following command to download
# ! wget https://repo1.maven.org/maven2/com/intel/analytics/bigdl/bigdl-spark_2.4.3/0.9.0/bigdl-spark_2.4.3-0.9.0-serving.jar
# if you are using wget to download, or get "bigdl-xxx-serving.jar" after "ls", please call mv *serving.jar bigdl.jar after downloaded.
# After initialization finished, check the directory
! ls
Explanation: Deploy Cluster Serving
After model prepared, we start to deploy it on Cluster Serving.
First install Cluster Serving
End of explanation
## BigDL Cluster Serving
model:
  # model path must be provided
  path: /tmp/transfer_learning_mobilenetv2
! head config.yaml
Explanation: We config the model path in config.yaml to following (the detail of config is at Cluster Serving Configuration)
End of explanation
! $FLINK_HOME/bin/start-cluster.sh
Explanation: Start Cluster Serving
Cluster Serving requires Flink and Redis installed, and corresponded environment variables set, check Cluster Serving Installation Guide for detail.
Flink cluster should start before Cluster Serving starts, if Flink cluster is not started, call following to start a local Flink cluster.
End of explanation
! cluster-serving-start
Explanation: After configuration, start Cluster Serving by cluster-serving-start (the detail is at Cluster Serving Programming Guide)
End of explanation
from bigdl.serving.client import InputQueue, OutputQueue
input_queue = InputQueue()
Explanation: Prediction using Cluster Serving
Next we start Cluster Serving code at python client.
End of explanation
arr = test_generator.next()[0]
arr
# Use async api to put and get, you have pass a name arg and use the name to get
input_queue.enqueue('my-input', t=arr)
output_queue = OutputQueue()
prediction = output_queue.query('my-input')
# Use sync api to predict, this will block until the result is get or timeout
prediction = input_queue.predict(arr)
prediction
Explanation: In Cluster Serving, only NdArray is supported as input. Thus, we first transform the generator to ndarray (If you do not know how to transform your input to NdArray, you may get help at data transform guide)
End of explanation
# don't forget to delete the model you save for this tutorial
! rm -rf /tmp/transfer_learning_mobilenetv2
Explanation: If everything works well, the result prediction would be the exactly the same NdArray object with the output of original Keras model.
End of explanation
<END_TASK>