学習済みモデルを用いて特徴量化（Azure Databricks）

# はじめに
これやっていきまーす

https://docs.databricks.com/_static/notebooks/deep-learning/deep-learning-transfer-learning-keras.html

# 開発環境

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/65ebbb62-0227-f143-6865-8d35ee44db53.png)

# 実装
1．ライブラリをインストール

~~~
%pip install tensorflow
~~~

2．ライブラリをインポート

~~~python
import pandas as pd
from PIL import Image
import numpy as np
import io

import tensorflow as tf
from tensorflow.keras.applications.resnet50 import ResNet50, preprocess_input
from tensorflow.keras.preprocessing.image import img_to_array

from pyspark.sql.functions import col, pandas_udf, PandasUDFType
~~~

3．flowers datasetを表示

~~~
%fs ls /databricks-datasets/flower_photos
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/2a02b4a4-fa13-ffda-6b5f-86dcdc51d5f5.png)

4．flowers datasetの読み込み

~~~python
images = spark.read.format("binaryFile") \
  .option("pathGlobFilter", "*.jpg") \
  .option("recursiveFileLookup", "true") \
  .load("/databricks-datasets/flower_photos")

display(images.limit(5))
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/6772bb74-e9c9-c115-1434-c323a4c9fb18.png)

5．モデルの準備（ResNet50）

~~~python
model = ResNet50(include_top=False)
model.summary()  # verify that the top layer is removed
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/a9514147-b189-aae7-4ef9-24faf875af88.png)

~~~python
bc_model_weights = sc.broadcast(model.get_weights())

def model_fn():
  """
  Returns a ResNet50 model with top layer removed and broadcasted pretrained weights.
  """
  model = ResNet50(weights=None, include_top=False)
  model.set_weights(bc_model_weights.value)
  return model
~~~

6．特徴量化

~~~python
def preprocess(content):
  """
  Preprocesses raw image bytes for prediction.
  """
  img = Image.open(io.BytesIO(content)).resize([224, 224])
  arr = img_to_array(img)
  return preprocess_input(arr)

def featurize_series(model, content_series):
  """
  Featurize a pd.Series of raw images using the input model.
  :return: a pd.Series of image features
  """
  input = np.stack(content_series.map(preprocess))
  preds = model.predict(input)
  # For some layers, output features will be multi-dimensional tensors.
  # We flatten the feature tensors to vectors for easier storage in Spark DataFrames.
  output = [p.flatten() for p in preds]
  return pd.Series(output)
~~~

~~~python
@pandas_udf('array<float>', PandasUDFType.SCALAR_ITER)
def featurize_udf(content_series_iter):
  '''
  This method is a Scalar Iterator pandas UDF wrapping our featurization function.
  The decorator specifies that this returns a Spark DataFrame column of type ArrayType(FloatType).
  
  :param content_series_iter: This argument is an iterator over batches of data, where each batch
                              is a pandas Series of image data.
  '''
  # With Scalar Iterator pandas UDFs, we can load the model once and then re-use it
  # for multiple data batches.  This amortizes the overhead of loading big models.
  model = model_fn()
  for content_series in content_series_iter:
    yield featurize_series(model, content_series)
~~~

~~~
/databricks/spark/python/pyspark/sql/pandas/functions.py:386: UserWarning: In Python 3.6+ and Spark 3.0+, it is preferred to specify type hints for pandas UDF instead of specifying pandas UDF type which will be deprecated in the future releases. See SPARK-28264 for more details.
  warnings.warn(
~~~

~~~python
# Pandas UDFs on large records (e.g., very large images) can run into Out Of Memory (OOM) errors.
# If you hit such errors in the cell below, try reducing the Arrow batch size via `maxRecordsPerBatch`.
spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", "1024")
~~~

~~~python
# We can now run featurization on our entire Spark DataFrame.
# NOTE: This can take a long time (about 10 minutes) since it applies a large model to the full dataset.
features_df = images.repartition(16).select(col("path"), featurize_udf("content").alias("features"))
features_df.write.mode("overwrite").parquet("dbfs:/ml/tmp/flower_photos_features")
~~~

7．この特徴量を用いて、新しいモデルを作って花の分類とかやってみよう。logistic regressionのサンプルとか見てね。

お疲れ様でした。