<jupyter_start><jupyter_text>Modules import<jupyter_code># Standard modules
import os
import io
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from PIL import Image
import boto3

# tensorflow modules
from tensorflow.keras.applications.resnet50 import ResNet50, preprocess_input
from tensorflow.keras.preprocessing.image import img_to_array

# pyspark modules
import pyspark 
from pyspark.sql import SparkSession
from pyspark import SparkContext, SparkConf
from pyspark.sql.functions import col, pandas_udf, PandasUDFType, split
from pyspark.ml.linalg import Vectors, VectorUDT
from pyspark.sql.functions import udf
from pyspark.ml.feature import StringIndexer, StandardScaler
from pyspark.ml.feature import PCA<jupyter_output><empty_output><jupyter_text>Spark initialization<jupyter_code># Instanciation of spark with credientials

spark = (SparkSession.builder        
         .appName('jv-p8')
         .config('spark.hadoop.fs.s3a.access.key', <access key>)
         .config('spark.hadoop.fs.s3a.secret.key', <secret key>)
         .config('spark.hadoop.fs.s3a.impl', 'org.apache.hadoop.fs.s3a.S3AFileSystem') 
         .getOrCreate()
        )

sc = spark.sparkContext
sc.setSystemProperty('com.amazonaws.services.s3a.enableV4', 'true')
sc._jsc.hadoopConfiguration().set("fs.s3a.endpoint", "s3.eu-west-1.amazonaws.com")<jupyter_output><empty_output><jupyter_text>Loading the data from the S3 bucket<jupyter_code># Vizualisation of the data

connection = boto3.client('s3')
contents = connection.list_objects(Bucket='s3-jv-ocr',
                                  Prefix='reduced_dataset/')['Contents']
for file in contents:
    print(file['Key'])
# Loading the data

s3_url = "s3a://s3-jv-ocr/reduced_dataset/*"
image_df = image_df = spark.read.format("binaryfile").load(s3_url)
image_df.printSchema()
# Data overview

image_df.show()
display(image_df)
# Data restructuring

image_df = image_df.withColumn('label', split(col('path'), '/').getItem(4))
image_df = image_df.select('path', 'content', 'label')
image_df.show()<jupyter_output><empty_output><jupyter_text>Model Downloading of the ResNet50 model file for featurization, and truncate the last layer(s).<jupyter_code># Deeplearning model instanciation

model = ResNet50(include_top=False)
model.summary()  # verify that the top layer is removed<jupyter_output><empty_output><jupyter_text>Functions<jupyter_code>bc_model_weights = spark.sparkContext.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
  
def preprocess(content):
    """
    Preprocesses raw image bytes for prediction.
    """
    img = Image.open(io.BytesIO(content)).resize([50, 50])
    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)
    output = [p.flatten() for p in preds]
    return pd.Series(output)
  
@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.
    '''
    model = model_fn()
    for content_series in content_series_iter:
        yield featurize_series(model, content_series)
# Application of functions
features_df = image_df.select(col("path"), col("label"), featurize_udf("content").alias("features"))
features_df.printSchema()
features_df.show()
# Modification of the type of data

features_vector = udf(lambda l: Vectors.dense(l), VectorUDT())
features_df = features_df.select(col("path"),  col("label"), features_vector(features_df["features"]).alias("features"))<jupyter_output><empty_output><jupyter_text>Dimensions reduction with a PCA<jupyter_code># standard scaler before doing the PCA

scaler = StandardScaler(withMean=True, withStd=True,
                              inputCol='features',
                              outputCol='features_scaled')
scaler_fitted = scaler.fit(features_df)
features_scaled_df = scaler_fitted.transform(features_df)
# I choose a number of dimensions pretty low because the cluster is on a single node to reduce the fees

pca = PCA(k=16, inputCol="features_scaled", outputCol="features_pca")
model = pca.fit(features_scaled_df)
df_transformed = model.transform(features_scaled_df)
df_transformed.show()<jupyter_output><empty_output><jupyter_text>Export the data scaled to S3<jupyter_code># Data selection and transformation to pandas

df_pandas = df_transformed.select('path', 'label', 'features_pca').toPandas()
# Export to S3 bucket

from io import StringIO

bucket = "s3-jv-ocr" 

csv_buffer = StringIO()
df_pandas.to_csv(csv_buffer)

s3_resource = boto3.resource('s3')

s3_resource.Object(bucket, 'df_reduced_pca_16.csv').put(Body=csv_buffer.getvalue())<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10631598444156325, 'fraction_numerical': 0.008890535284311909, 'mean_word_length': 4.806451612903226, 'pattern_counts': {'":': 0, '<': 32, '<?xml version=': 0, '>': 32, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '245965', 'license': 'Unlicense', 'n_tokens_mistral': 1792, 'n_tokens_neox': 1744, 'n_words': 399, 'path': 'notebooks/ds_p8.ipynb', 'repo_name': 'j30v/j30v.github.io', 'size': '5399'}