{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!pip install numpy==1.22.2","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:44:17.544064Z","iopub.execute_input":"2023-08-04T17:44:17.544578Z","iopub.status.idle":"2023-08-04T17:44:39.231871Z","shell.execute_reply.started":"2023-08-04T17:44:17.544535Z","shell.execute_reply":"2023-08-04T17:44:39.230655Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\nimport os\nimport glob\nimport re\nfrom IPython.display import display\nfrom decimal import Decimal\nimport timeit\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2023-08-04T17:44:39.233389Z","iopub.execute_input":"2023-08-04T17:44:39.233775Z","iopub.status.idle":"2023-08-04T17:44:39.253015Z","shell.execute_reply.started":"2023-08-04T17:44:39.233732Z","shell.execute_reply":"2023-08-04T17:44:39.251746Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!pip install pyspark","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:44:39.255093Z","iopub.execute_input":"2023-08-04T17:44:39.255567Z","iopub.status.idle":"2023-08-04T17:45:35.226852Z","shell.execute_reply.started":"2023-08-04T17:44:39.255523Z","shell.execute_reply":"2023-08-04T17:45:35.225316Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pyspark.sql import functions as F\nfrom pyspark.sql import SparkSession, DataFrame\nfrom pyspark.sql.types import IntegerType, StringType, FloatType, ArrayType, StructType, StructField, DoubleType\nfrom pyspark.sql.functions import col, avg, sum, split, rand, datediff, lit, abs, collect_list, count, udf, concat_ws\nfrom pyspark.sql.window import Window\nfrom pyspark.ml import Pipeline\nfrom pyspark.ml.recommendation import ALS\nfrom pyspark.ml.evaluation import RegressionEvaluator\nfrom pyspark.ml.tuning import TrainValidationSplit, ParamGridBuilder\nfrom pyspark.ml.stat import Summarizer\nfrom pyspark.ml.feature import VectorAssembler, StringIndexer, OneHotEncoder, MinHashLSH\nfrom pyspark.ml.linalg import Vectors, VectorUDT","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.230918Z","iopub.execute_input":"2023-08-04T17:45:35.231331Z","iopub.status.idle":"2023-08-04T17:45:35.449814Z","shell.execute_reply.started":"2023-08-04T17:45:35.231298Z","shell.execute_reply":"2023-08-04T17:45:35.448554Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def definir_directorios():\n    DIRECTORY = \"/kaggle/input/datathon2023\"\n    OUTPUT_DIRECTORY = \"/kaggle/working\"\n    return DIRECTORY, OUTPUT_DIRECTORY","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.451659Z","iopub.execute_input":"2023-08-04T17:45:35.452260Z","iopub.status.idle":"2023-08-04T17:45:35.457727Z","shell.execute_reply.started":"2023-08-04T17:45:35.452224Z","shell.execute_reply":"2023-08-04T17:45:35.456325Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def check_nulls_in_dataframe(df):\n    # Check for null values in each column\n    null_counts = df.select([F.count(F.when(F.col(c).isNull(), c)).alias(c) for c in df.columns])\n    null_counts.show()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def calculate_array_length(arr):\n    return len(arr)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Definir una función definida por el usuario (UDF) para convertir vectores densos en vectores dispersos\ndef dense_to_sparse(v):\n    return Vectors.sparse(v.size, v.indices, v.values)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_product_data(spark, directory):\n    return spark.read.csv(\n        os.path.join(directory, \"RentabilidadProduct.csv\"),\n        sep=\"|\",\n        header=True,\n        inferSchema=True\n    )","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_train_data(spark, directory):\n    return spark.read.csv(\n        os.path.join(directory, \"train.csv\"),\n        sep=\"|\",\n        header=True,\n        inferSchema=True,\n        timestampFormat=\"yyyy-MM-dd\",\n    )","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_comprados_from_train_data(spark, train_data):\n    # Drop a las columnas no necesarias para esta parte del programa\n    columns_to_drop = [\"fecha_compra\", \"type_id\", \"business_id\", \"channel_id\", \"monto_venta_transf\", \"peso_venta_transf\"]\n    \n    # return comprados_from_train_data\n    return (train_data\n                .drop(*columns_to_drop)\n                .groupBy(\"customer_id\", \"product_id\")\n                .agg(F.sum(\"cantidad_venta\").alias(\"acum_cantidad_venta\"))\n                .orderBy(\"customer_id\", \"product_id\")\n    )","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.460158Z","iopub.execute_input":"2023-08-04T17:45:35.460901Z","iopub.status.idle":"2023-08-04T17:45:35.476141Z","shell.execute_reply.started":"2023-08-04T17:45:35.460862Z","shell.execute_reply":"2023-08-04T17:45:35.474637Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_recomendados_from_submission_data(spark, submission_data):\n    return (\n            submission_data\n                .withColumn('product_id', F.split(submission_data['product_id'], ' '))\n                .withColumn('product_id', F.slice('product_id', 1, 30))\n                .select(\"customer_id\", F.explode(\"product_id\").alias(\"product_id\")) # Explode product_id en filas individuales\n                .withColumn(\"product_id\", F.col(\"product_id\").cast(\"int\")) # Convertir de string a integer\n                .orderBy(\"customer_id\", \"product_id\") # Ordenar\n        )\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.477914Z","iopub.execute_input":"2023-08-04T17:45:35.478354Z","iopub.status.idle":"2023-08-04T17:45:35.500532Z","shell.execute_reply.started":"2023-08-04T17:45:35.478319Z","shell.execute_reply":"2023-08-04T17:45:35.499361Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_cantidad_comprados_y_recomendados(spark, comprados, recomendados):\n    # Realizar inner join en 'customer_id' y 'product_id'\n    comprados_por_customer = (\n        comprados\n            .groupBy('customer_id')\n            .agg(F.sum('acum_cantidad_venta').alias('sum_cantidad_comprados'))\n    )\n    \n    comprados_y_recomendados_por_customer = (\n        comprados\n            .join(recomendados, ['customer_id', 'product_id'], \"inner\")\n            .groupBy('customer_id')\n            .agg(F.sum('acum_cantidad_venta').alias('sum_cantidad_comprados_y_recomendados'))\n    )\n\n    return comprados_y_recomendados_por_customer, comprados_por_customer\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.501996Z","iopub.execute_input":"2023-08-04T17:45:35.502436Z","iopub.status.idle":"2023-08-04T17:45:35.517669Z","shell.execute_reply.started":"2023-08-04T17:45:35.502403Z","shell.execute_reply":"2023-08-04T17:45:35.516024Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_recall_por_cliente(spark, comprados_por_customer, comprados_y_recomendados_por_customer):\n    # Hacemos un left join para fusionar los dos dataframes sin perder data de cantidad_comprados\n    return (\n        comprados_por_customer\n            .join(comprados_y_recomendados_por_customer, on='customer_id', how='left')\n            .orderBy(\"customer_id\")\n            # Reemplazar null values en 'sum_cantidad_comprados_y_recomendados' con 0\n            .fillna(0, subset=['sum_cantidad_comprados_y_recomendados'])\n            # Añadimos una columna \"recall\"\n            .withColumn(\"recall\", F.col(\"sum_cantidad_comprados_y_recomendados\") / F.col(\"sum_cantidad_comprados\"))\n            .orderBy(\"customer_id\")\n    )","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.520027Z","iopub.execute_input":"2023-08-04T17:45:35.521460Z","iopub.status.idle":"2023-08-04T17:45:35.540708Z","shell.execute_reply.started":"2023-08-04T17:45:35.521402Z","shell.execute_reply":"2023-08-04T17:45:35.539016Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def call_funcion_get_recall_por_cliente(spark, directory, train_data, submission_data):\n    # Generar comprados_data a partir de train.csv\n    comprados_data = get_comprados_from_train_data(spark, train_data)\n\n    # Extraer recomendados a partir de un submission previo\n    recomendados_data = get_recomendados_from_submission_data(spark, submission_data)\n\n    # Extraer comprado y recomendados, y comprados \n    comprados_y_recomendados_por_customer, comprados_por_customer = get_cantidad_comprados_y_recomendados(spark, comprados_data, recomendados_data)\n\n    # Llamar función get_recall_por_cliente\n    return get_recall_por_cliente(spark, comprados_por_customer, comprados_y_recomendados_por_customer)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.542684Z","iopub.execute_input":"2023-08-04T17:45:35.543158Z","iopub.status.idle":"2023-08-04T17:45:35.558852Z","shell.execute_reply.started":"2023-08-04T17:45:35.543122Z","shell.execute_reply":"2023-08-04T17:45:35.557408Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def completar_SKUS_en_product_data(spark, train_data, product_data):\n    # Encontrar los product_ids faltantes en product_data\n    existing_product_ids = product_data.select(\"product_id\").distinct()\n    train_product_ids = train_data.select(\"product_id\").distinct()\n    missing_product_ids = train_product_ids.subtract(existing_product_ids)\n\n    # Calcular el average de utilidad_bruta_transf en product_data\n    average_utilidad_bruta_transf = product_data.agg(avg(col(\"utilidad_bruta_transf\"))).collect()[0][0]\n\n    # Añadir los product_ids faltantes a product_data con el average de utilidad_bruta_transf\n    average_utilidad_bruta_transf_decimal = Decimal(str(average_utilidad_bruta_transf))\n    missing_product_data = missing_product_ids.withColumn(\"utilidad_bruta_transf\", lit(average_utilidad_bruta_transf_decimal))\n\n    # Add missing columns to missing_product_data and fill them with zeros\n    missing_product_data = missing_product_data.withColumn(\"family_id\", lit(0))\n    missing_product_data = missing_product_data.withColumn(\"negocio_id\", lit(0))\n    missing_product_data = missing_product_data.withColumn(\"category_id\", lit(0))\n    missing_product_data = missing_product_data.withColumn(\"tier_id\", lit(0))\n\n    # Union product_data with missing_product_data ==> product_data_con_SKUs_completos\n    return product_data.union(missing_product_data)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.603248Z","iopub.execute_input":"2023-08-04T17:45:35.604863Z","iopub.status.idle":"2023-08-04T17:45:35.615052Z","shell.execute_reply.started":"2023-08-04T17:45:35.604799Z","shell.execute_reply":"2023-08-04T17:45:35.613975Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_train_data_ready_for_model(spark, train_data, product_data_con_SKUs_completos):\n    categorical_columns = [\"business_id\", \"type_id\", \"channel_id\"]\n    product_data_sin_product_id = product_data_con_SKUs_completos.drop(\"product_id\")\n    \n    # Fusión de train_data con product_data para obtener la utilidad de cada producto\n    final_df = train_data.join(product_data_con_SKUs_completos, on=\"product_id\", how=\"left\") \\\n        .withColumn(\"dias_desde_ultima_compra\", datediff(lit(\"2023-02-05\"), F.col(\"fecha_compra\"))) \\\n        .withColumn(\"combined_rating\", (F.col(\"utilidad_bruta_transf\") * F.col(\"cantidad_venta\")  / (1 + F.col(\"dias_desde_ultima_compra\"))))\n    \n    final_df = final_df.fillna({'utilidad_bruta_transf': 0, 'combined_rating': 0, 'dias_desde_ultima_compra': 372})\n    \n    for column in categorical_columns:\n        final_df = final_df.withColumn(column, F.col(column).cast(StringType()).alias(column))\n    \n    final_df = final_df.select(\n        *train_data.columns,\n        \"dias_desde_ultima_compra\",\n        \"combined_rating\",\n        *product_data_sin_product_id.columns\n    ).drop(\"fecha_compra\")\n    \n    return final_df\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.617121Z","iopub.execute_input":"2023-08-04T17:45:35.618145Z","iopub.status.idle":"2023-08-04T17:45:35.636866Z","shell.execute_reply.started":"2023-08-04T17:45:35.618096Z","shell.execute_reply":"2023-08-04T17:45:35.635532Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Función para generar Item Profiles:\ndef generate_perfiles_de_productos(spark, dataframe, categorical_cols):\n    # Drop filas con valores nulos\n    df_with_selected_attributes = dataframe.na.drop(subset=categorical_cols)\n\n    # Indexar columnas categóricas definidas explícitamente para convertirlas en representaciones numéricas\n    indexers = [StringIndexer(inputCol=col_name, outputCol=f\"{col_name}_index\") for col_name in categorical_cols]\n    pipeline = Pipeline(stages=indexers)\n    df_with_indexed_attributes = pipeline.fit(df_with_selected_attributes).transform(df_with_selected_attributes)\n\n    # OneHotEncode las columnas categóricas indexadas para convertirlas en vectores binarios\n    encoders = [OneHotEncoder(inputCol=f\"{col_name}_index\", outputCol=f\"{col_name}_encoded\") for col_name in categorical_cols]\n    pipeline = Pipeline(stages=encoders)\n    df_with_encoded_attributes = pipeline.fit(df_with_indexed_attributes).transform(df_with_indexed_attributes)\n\n    # Reunir los atributos seleccionados en una columna de vectores de características\n    input_cols = [f\"{col_name}_encoded\" for col_name in categorical_cols]\n    assembler = VectorAssembler(inputCols=input_cols, outputCol=\"item_profile\")\n\n    # return product_profiles_df\n    return assembler.transform(df_with_encoded_attributes).select(\"product_id\", \"item_profile\")\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.638445Z","iopub.execute_input":"2023-08-04T17:45:35.638978Z","iopub.status.idle":"2023-08-04T17:45:35.664435Z","shell.execute_reply.started":"2023-08-04T17:45:35.638925Z","shell.execute_reply":"2023-08-04T17:45:35.662977Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Función para generar User Profiles (perfiles de clientes):\ndef generate_user_profiles(spark, dataframe, user_col, item_col):\n    # return customer_profiles_df\n    return (\n        # Find the user's purchased products\n        dataframe.filter(col(\"cantidad_venta\") > 0).select(user_col, item_col).distinct()\n        # Calculate user preferences (aggregating item interactions)\n        .groupBy(user_col).agg(\n            collect_list(item_col).alias(\"user_interactions\"),\n            count(item_col).alias(\"interactions_count\")\n        )\n    )\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.666954Z","iopub.execute_input":"2023-08-04T17:45:35.668244Z","iopub.status.idle":"2023-08-04T17:45:35.686519Z","shell.execute_reply.started":"2023-08-04T17:45:35.668178Z","shell.execute_reply":"2023-08-04T17:45:35.685095Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Función para generar Item Similarities\ndef generate_product_similarities(spark, product_profiles_df):\n    # Convierte los perfiles densos de los elementos en vectores dispersos (necesario para MinHashLSH)\n    sparse_item_profiles_df = product_profiles_df.withColumn(\"item_profile_sparse\", to_sparse_udf(\"item_profile\"))\n\n    # Crear modelo MinHashLSH\n    num_hash_tables = 10\n    min_hash_lsh = MinHashLSH(inputCol=\"item_profile_sparse\", outputCol=\"hashes\", numHashTables=num_hash_tables)\n    model = min_hash_lsh.fit(sparse_item_profiles_df)\n    \n    # Incrementar el threshold para aproximar la similaridad default: 0.8\n    threshold = 0.5\n    \n    # return dataframe product_similarities_df usando MinHashLSH\n    product_similarities_df =  model.approxSimilarityJoin(sparse_item_profiles_df, sparse_item_profiles_df, threshold, distCol=\"similarity\")\n    \n    return product_similarities_df\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.706603Z","iopub.execute_input":"2023-08-04T17:45:35.707165Z","iopub.status.idle":"2023-08-04T17:45:35.726627Z","shell.execute_reply.started":"2023-08-04T17:45:35.707112Z","shell.execute_reply":"2023-08-04T17:45:35.725472Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def generate_customer_purchases(spark, train_data):\n    # Partimos del DataFrame \"train_data\" que contiene las columnas \"customer_id\" y \"product_id\" para las interacciones usuario-artículo (compras).\n    # Este DataFrame se utilizará para filtrar el item_similarities_df\n\n    # Encontrar los productos comprados por el customer\n    customer_purchases_df = train_data.filter(F.col(\"cantidad_venta\") > 0).select(\"customer_id\", \"product_id\").distinct()\n        \n    return customer_purchases_df\n    ","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.746164Z","iopub.execute_input":"2023-08-04T17:45:35.747529Z","iopub.status.idle":"2023-08-04T17:45:35.760481Z","shell.execute_reply.started":"2023-08-04T17:45:35.747471Z","shell.execute_reply":"2023-08-04T17:45:35.758800Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def generate_grouped_recommendations(spark, customer_purchases_df, product_similarities_df):\n    try:\n        # Filtrar item (product) similarities para cada user (customer) basados en sus productos comprados\n        window_spec = Window.partitionBy(\"user_id\").orderBy(F.col(\"similarity\").desc())\n\n        top_30_recommendations_df = (\n            product_similarities_df.alias(\"similarity_df\")\n                .join(\n                    customer_purchases_df.alias(\"purchases_df\"),\n                    F.col(\"similarity_df.datasetA.product_id\") == F.col(\"purchases_df.product_id\"),\n                    \"inner\"\n                )\n        )\n\n        # Before creating product_id_list and rank_list\n        print(\"Before creating rank\")\n        # Print schema and sample data of top_30_recommendations_df\n        print(\"Schema and Sample Data of top_30_recommendations_df:\")\n        top_30_recommendations_df.printSchema()\n\n        top_30_recommendations_df = (\n            top_30_recommendations_df.select(\n                    F.col(\"purchases_df.customer_id\").alias(\"user_id\"),\n                    F.col(\"similarity_df.datasetB.product_id\").alias(\"recommended_product_id\"),\n                    F.col(\"similarity_df.similarity\")\n                )\n                # Clasificar los productos recomendados para cada usuario en función del ranking de similitud\n                .withColumn(\"rank\", F.dense_rank().over(window_spec))\n                # Recopilar las 30 recomendaciones mejor clasificadas para cada usuario\n                .where(F.col(\"rank\") <= 30)\n        )\n\n\n        # Before creating product_id_list and rank_list\n        print(\"Before creating product_id_list and rank_list\")\n        # Print schema and sample data of top_30_recommendations_df\n        print(\"Schema and Sample Data of top_30_recommendations_df:\")\n        top_30_recommendations_df.printSchema()\n\n        grouped_recommendations_df = (\n            top_30_recommendations_df\n                .groupby(\"user_id\")\n                .agg(\n                    F.collect_list(\"recommended_product_id\").alias(\"product_id_list\"),\n                    F.collect_list(\"rank\").alias(\"rank_list\")\n                )\n                # Sort the DataFrame by rank within each group (user_id)\n                .withColumn(\n                    \"sorted_recommended_products\",\n                    F.array_sort(\n                        F.arrays_zip(F.col(\"rank_list\"), F.col(\"product_id_list\"))\n                    )\n                )\n                .select(\n                    \"user_id\",\n                    F.col(\"sorted_recommended_products.product_id_list\").alias(\"product_id_list\"),\n                    F.col(\"sorted_recommended_products.rank_list\").alias(\"rank_list\")\n                )\n                .drop(\"sorted_recommended_products\")\n        )\n        \n    except Exception as e:\n        print(\"An error occurred:\", e)\n        return None  # Return None or handle the error in an appropriate way\n    \n    return grouped_recommendations_df\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.762540Z","iopub.execute_input":"2023-08-04T17:45:35.762916Z","iopub.status.idle":"2023-08-04T17:45:35.778578Z","shell.execute_reply.started":"2023-08-04T17:45:35.762886Z","shell.execute_reply":"2023-08-04T17:45:35.777123Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def manual_stratified_split(spark, dataframe, stratify_col, train_ratio=0.8, seed=42):\n    # Calcular count de total de filas y distinct customer_ids\n    # total_row_count = dataframe.count()\n    total_row_count = 2167848\n    # distinct_customer_count = dataframe.select(stratify_col).distinct().count()\n    distinct_customer_count = 15312\n\n    # Calcular el número de filas a muestrear por customer_id\n    num_train_rows_per_customer = int(train_ratio * total_row_count / distinct_customer_count)\n    \n    # Crear un mapeo de customer_id a sus filas correspondientes\n    customer_rows_map = dataframe.groupBy(stratify_col).agg(F.collect_list(F.struct(*dataframe.columns)).alias(\"rows\"))\n\n    # Muestrear filas de cada grupo customer_id proporcionalmente a train_ratio\n    sampled_rows_per_customer = customer_rows_map.withColumn(\"sampled_rows\", F.expr(f\"slice(rows, 1, {num_train_rows_per_customer})\"))\n\n    # Crear DataFrames de training_data y validation_data\n    training_data = sampled_rows_per_customer.select(F.explode(\"sampled_rows\").alias(\"sampled_row\")).select(\"sampled_row.*\")\n    validation_data = dataframe.subtract(training_data)\n\n    return training_data, validation_data","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.780537Z","iopub.execute_input":"2023-08-04T17:45:35.780931Z","iopub.status.idle":"2023-08-04T17:45:35.799423Z","shell.execute_reply.started":"2023-08-04T17:45:35.780899Z","shell.execute_reply":"2023-08-04T17:45:35.798265Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def build_model_ALS(spark, training_data):\n    # Construir modelo ALS con la columna combined_rating como rating\n    return ALS(\n        rank=10,         # Número de dimensiones latentes\n        maxIter=10,      # Número máximo de iteraciones (10 es el default)\n        regParam=0.1,    # Parámetro de regularización\n        userCol=\"customer_id\",\n        itemCol=\"product_id\",\n        ratingCol=\"combined_rating\",   # Utilizaremos combined_rating como valor de rating\n    )","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.800871Z","iopub.execute_input":"2023-08-04T17:45:35.801770Z","iopub.status.idle":"2023-08-04T17:45:35.819675Z","shell.execute_reply.started":"2023-08-04T17:45:35.801703Z","shell.execute_reply":"2023-08-04T17:45:35.818724Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Definir una función UDF para comprimir product_id y rank_lists, y ordenarlos por rank\n@F.udf(returnType=F.StringType())\ndef zip_and_sort(product_id_list, rank_list):\n    sorted_pairs = sorted(zip(product_id_list, rank_list), key=lambda x: x[1])\n    sorted_product_ids = [str(pair[0]) for pair in sorted_pairs]\n    return \" \".join(sorted_product_ids)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.820917Z","iopub.execute_input":"2023-08-04T17:45:35.822180Z","iopub.status.idle":"2023-08-04T17:45:35.836475Z","shell.execute_reply.started":"2023-08-04T17:45:35.822137Z","shell.execute_reply":"2023-08-04T17:45:35.835199Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def generar_30_SKUs_recomendados(spark, train_data_ready_for_model, model):\n    # Generar recomendaciones para los clientes en la tabla de sumisión\n    # submission_data = train_data_ready_for_model.select(\"customer_id\").distinct()  # Usa train_data_ready_for_model para generar recomendaciones\n    submission_data = train_data_ready_for_model.groupBy(\"customer_id\").agg(F.first(\"customer_id\").alias(\"customer_id\"))\n\n    recommendations = model.recommendForUserSubset(submission_data, 30)  # Genera top 30 recomendaciones\n\n    recommendations_df = (\n        recommendations\n            # Convertir el resultado en un DataFrame\n            .selectExpr(\"customer_id\", \"explode(recommendations) as recommendation\")\n            # Extraer 'product_id' y 'rating' de la columna recommendation\n            .select(\"customer_id\", F.col(\"recommendation.product_id\"), F.col(\"recommendation.rating\"))\n            # Agregar el rank de los productos recomendados\n            .withColumn(\"rank\", F.row_number().over(Window.partitionBy(\"customer_id\").orderBy(F.desc(\"rating\"))))\n    )\n    \n    submission_df = (\n        recommendations_df\n            # Agrupar el resultado por customer_id y collect product_id y rank en listas\n            .groupBy(\"customer_id\")\n            .agg(\n                F.collect_list(\"product_id\").alias(\"product_id_list\"),\n                F.collect_list(\"rank\").alias(\"rank_list\")\n            )\n    )\n    \n    return submission_df\n\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.837918Z","iopub.execute_input":"2023-08-04T17:45:35.838390Z","iopub.status.idle":"2023-08-04T17:45:35.856701Z","shell.execute_reply.started":"2023-08-04T17:45:35.838360Z","shell.execute_reply":"2023-08-04T17:45:35.855378Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def save_recommendations_to_submission(spark, reformatted_hybrid_submision_df, output_directory):\n    # Guardar el DataFrame reformatted_hybrid_submision_df como CSV en el directorio '/kaggle/working'\n    # Encontrar el último archivo de submission\n    existing_files = glob.glob(os.path.join(output_directory, \"submission*.csv\"))\n\n    # Extraer la parte numérica de los nombres de los archivos encontradis usando regex\n    file_numbers = [int(re.findall(r'\\d+', file)[-1]) for file in existing_files]\n\n    if file_numbers:\n        last_file_number = max(file_numbers)\n        next_file_number = last_file_number + 1\n    else:\n        next_file_number = 1\n\n    reformatted_hybrid_submision_df = reformatted_hybrid_submision_df.repartition(200)  # Adjust the number of partitions as needed\n\n    # Crear el nuevo nombre del file submission\n    submission_filename = os.path.join(output_directory, f\"submission{next_file_number:03d}.csv\")\n    \n    print(\"submission_filename antes de convertir a csv: \", submission_filename)\n    \n    # reformatted_hybrid_submision_df.toPandas().to_csv(submission_filename, header=True, index=False)\n    \n    # Convertir el DataFrame directamente a CSV usando Spark\n    (\n        reformatted_hybrid_submision_df\n            .coalesce(1)  # Coalesce to a single partition\n            .write\n            .csv(submission_filename, header=True, mode=\"overwrite\")\n    )\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.858928Z","iopub.execute_input":"2023-08-04T17:45:35.859584Z","iopub.status.idle":"2023-08-04T17:45:35.874765Z","shell.execute_reply.started":"2023-08-04T17:45:35.859550Z","shell.execute_reply":"2023-08-04T17:45:35.873552Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Iniciar Spark session\nspark = (SparkSession\n            .builder\n            .master(\"local\")\n            .appName(\"optimizing SKUs\")\n            .config(\"spark.executor.memory\", \"8g\")\n            .config(\"spark.driver.memory\", \"8g\")\n            .config(\"spark.executor.cores\", \"8\")\n            .config(\"spark.sql.shuffle.partitions\", \"100\")  # Adjust the number of partitions as needed\n            # .config(\"spark.driver.extraLibraryPath\", \"/usr/lib/x86_64-linux-gnu/blas:/usr/lib/x86_64-linux-gnu/lapack\")\n            # .config(\"spark.executor.extraLibraryPath\", \"/usr/lib/x86_64-linux-gnu/blas:/usr/lib/x86_64-linux-gnu/lapack\")\n            .getOrCreate()\n    )\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:35.876563Z","iopub.execute_input":"2023-08-04T17:45:35.876909Z","iopub.status.idle":"2023-08-04T17:45:42.668295Z","shell.execute_reply.started":"2023-08-04T17:45:35.876879Z","shell.execute_reply":"2023-08-04T17:45:42.667153Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Registrar UDF\nto_sparse_udf = udf(dense_to_sparse, VectorUDT())\nspark.udf.register(\"to_sparse_udf\", to_sparse_udf)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:42.669700Z","iopub.execute_input":"2023-08-04T17:45:42.670067Z","iopub.status.idle":"2023-08-04T17:45:44.127848Z","shell.execute_reply.started":"2023-08-04T17:45:42.670011Z","shell.execute_reply":"2023-08-04T17:45:44.126580Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Definir directorios de entrada y salida \nDIRECTORY, OUTPUT_DIRECTORY = definir_directorios()\nDIRECTORY, OUTPUT_DIRECTORY","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:44.130241Z","iopub.execute_input":"2023-08-04T17:45:44.131222Z","iopub.status.idle":"2023-08-04T17:45:44.145283Z","shell.execute_reply.started":"2023-08-04T17:45:44.131171Z","shell.execute_reply":"2023-08-04T17:45:44.143302Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Obtener dataframe train_data\ntrain_data = get_train_data(spark, DIRECTORY)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:45:44.147414Z","iopub.execute_input":"2023-08-04T17:45:44.148624Z","iopub.status.idle":"2023-08-04T17:46:05.140255Z","shell.execute_reply.started":"2023-08-04T17:45:44.148558Z","shell.execute_reply":"2023-08-04T17:46:05.137819Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_data.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:05.141501Z","iopub.execute_input":"2023-08-04T17:46:05.141858Z","iopub.status.idle":"2023-08-04T17:46:05.234460Z","shell.execute_reply.started":"2023-08-04T17:46:05.141828Z","shell.execute_reply":"2023-08-04T17:46:05.233137Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_data.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:05.235891Z","iopub.execute_input":"2023-08-04T17:46:05.236382Z","iopub.status.idle":"2023-08-04T17:46:05.531145Z","shell.execute_reply.started":"2023-08-04T17:46:05.236344Z","shell.execute_reply":"2023-08-04T17:46:05.530098Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_data.show()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:05.532721Z","iopub.execute_input":"2023-08-04T17:46:05.533285Z","iopub.status.idle":"2023-08-04T17:46:23.108969Z","shell.execute_reply.started":"2023-08-04T17:46:05.533251Z","shell.execute_reply":"2023-08-04T17:46:23.107767Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"check_nulls_in_dataframe(train_data)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:23.110345Z","iopub.execute_input":"2023-08-04T17:46:23.110763Z","iopub.status.idle":"2023-08-04T17:46:28.392132Z","shell.execute_reply.started":"2023-08-04T17:46:23.110725Z","shell.execute_reply":"2023-08-04T17:46:28.390695Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Obtener dataframe product_data\nproduct_data = get_product_data(spark, DIRECTORY)\nproduct_data.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:28.393673Z","iopub.execute_input":"2023-08-04T17:46:28.394329Z","iopub.status.idle":"2023-08-04T17:46:28.894841Z","shell.execute_reply.started":"2023-08-04T17:46:28.394280Z","shell.execute_reply":"2023-08-04T17:46:28.893735Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"product_data\")\nproduct_data.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:28.896120Z","iopub.execute_input":"2023-08-04T17:46:28.896513Z","iopub.status.idle":"2023-08-04T17:46:28.904625Z","shell.execute_reply.started":"2023-08-04T17:46:28.896477Z","shell.execute_reply":"2023-08-04T17:46:28.903454Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_data.count()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:28.906240Z","iopub.execute_input":"2023-08-04T17:46:28.906979Z","iopub.status.idle":"2023-08-04T17:46:29.279615Z","shell.execute_reply.started":"2023-08-04T17:46:28.906935Z","shell.execute_reply":"2023-08-04T17:46:29.278614Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_data.show()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:29.283557Z","iopub.execute_input":"2023-08-04T17:46:29.284950Z","iopub.status.idle":"2023-08-04T17:46:29.489645Z","shell.execute_reply.started":"2023-08-04T17:46:29.284851Z","shell.execute_reply":"2023-08-04T17:46:29.488243Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"check_nulls_in_dataframe(product_data)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:29.491144Z","iopub.execute_input":"2023-08-04T17:46:29.491591Z","iopub.status.idle":"2023-08-04T17:46:29.927233Z","shell.execute_reply.started":"2023-08-04T17:46:29.491549Z","shell.execute_reply":"2023-08-04T17:46:29.925719Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Completar product_data con los product_id adicionales que se encuentran en train_data\nproduct_data_con_SKUs_completos = completar_SKUS_en_product_data(spark, train_data, product_data)\nproduct_data_con_SKUs_completos.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:29.941537Z","iopub.execute_input":"2023-08-04T17:46:29.943457Z","iopub.status.idle":"2023-08-04T17:46:30.669051Z","shell.execute_reply.started":"2023-08-04T17:46:29.943397Z","shell.execute_reply":"2023-08-04T17:46:30.668122Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"product_data_con_SKUs_completos\")\nproduct_data_con_SKUs_completos.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:30.673464Z","iopub.execute_input":"2023-08-04T17:46:30.675725Z","iopub.status.idle":"2023-08-04T17:46:30.683606Z","shell.execute_reply.started":"2023-08-04T17:46:30.675684Z","shell.execute_reply":"2023-08-04T17:46:30.682459Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_data_con_SKUs_completos.count()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:30.686238Z","iopub.execute_input":"2023-08-04T17:46:30.687252Z","iopub.status.idle":"2023-08-04T17:46:40.168200Z","shell.execute_reply.started":"2023-08-04T17:46:30.687207Z","shell.execute_reply":"2023-08-04T17:46:40.166900Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_data_con_SKUs_completos.show()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:40.170382Z","iopub.execute_input":"2023-08-04T17:46:40.170866Z","iopub.status.idle":"2023-08-04T17:46:40.367336Z","shell.execute_reply.started":"2023-08-04T17:46:40.170823Z","shell.execute_reply":"2023-08-04T17:46:40.365900Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"check_nulls_in_dataframe(product_data_con_SKUs_completos)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:40.368763Z","iopub.execute_input":"2023-08-04T17:46:40.369228Z","iopub.status.idle":"2023-08-04T17:46:43.121755Z","shell.execute_reply.started":"2023-08-04T17:46:40.369190Z","shell.execute_reply":"2023-08-04T17:46:43.120170Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Preparar data para ser entrenada con Model\ntrain_data_ready_for_model = get_train_data_ready_for_model(spark, train_data, product_data_con_SKUs_completos)\ntrain_data_ready_for_model.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:43.123343Z","iopub.execute_input":"2023-08-04T17:46:43.123836Z","iopub.status.idle":"2023-08-04T17:46:43.638836Z","shell.execute_reply.started":"2023-08-04T17:46:43.123790Z","shell.execute_reply":"2023-08-04T17:46:43.637464Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"train_data_ready_for_model\")\ntrain_data_ready_for_model.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:43.640390Z","iopub.execute_input":"2023-08-04T17:46:43.640856Z","iopub.status.idle":"2023-08-04T17:46:43.650482Z","shell.execute_reply.started":"2023-08-04T17:46:43.640814Z","shell.execute_reply":"2023-08-04T17:46:43.649107Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_data_ready_for_model.count()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:43.653059Z","iopub.execute_input":"2023-08-04T17:46:43.653995Z","iopub.status.idle":"2023-08-04T17:46:59.399632Z","shell.execute_reply.started":"2023-08-04T17:46:43.653951Z","shell.execute_reply":"2023-08-04T17:46:59.398795Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_data_ready_for_model.show()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:59.400557Z","iopub.execute_input":"2023-08-04T17:46:59.400854Z","iopub.status.idle":"2023-08-04T17:46:59.592598Z","shell.execute_reply.started":"2023-08-04T17:46:59.400827Z","shell.execute_reply":"2023-08-04T17:46:59.591146Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"check_nulls_in_dataframe(train_data_ready_for_model)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:46:59.594069Z","iopub.execute_input":"2023-08-04T17:46:59.594534Z","iopub.status.idle":"2023-08-04T17:47:01.793761Z","shell.execute_reply.started":"2023-08-04T17:46:59.594490Z","shell.execute_reply":"2023-08-04T17:47:01.792622Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Generar Perfiles de Usuarios (por preferencias) y Perfiles de Productos (por similaridades de ítems)\n# Definir selected attributes para item profiles\nselected_attributes = [\"type_id\", \"business_id\", \"channel_id\", \"family_id\", \"negocio_id\", \"category_id\", \"tier_id\"]\n\ntrain_data_ready_for_model_repartitioned = train_data_ready_for_model.repartition(100)\n\n# Generar los profiles de productos\nproduct_profiles_df = generate_perfiles_de_productos(spark, train_data_ready_for_model_repartitioned, selected_attributes)\nproduct_profiles_df.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:47:01.795171Z","iopub.execute_input":"2023-08-04T17:47:01.795624Z","iopub.status.idle":"2023-08-04T17:47:47.394902Z","shell.execute_reply.started":"2023-08-04T17:47:01.795582Z","shell.execute_reply":"2023-08-04T17:47:47.393569Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"product_profiles_df\")\nproduct_profiles_df.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:47:47.396676Z","iopub.execute_input":"2023-08-04T17:47:47.397188Z","iopub.status.idle":"2023-08-04T17:47:47.405955Z","shell.execute_reply.started":"2023-08-04T17:47:47.397143Z","shell.execute_reply":"2023-08-04T17:47:47.404254Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_profiles_df.repartition(100).count()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:47:47.409198Z","iopub.execute_input":"2023-08-04T17:47:47.410267Z","iopub.status.idle":"2023-08-04T17:48:25.496371Z","shell.execute_reply.started":"2023-08-04T17:47:47.410178Z","shell.execute_reply":"2023-08-04T17:48:25.494735Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_profiles_df.show(truncate=False)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:48:25.498492Z","iopub.execute_input":"2023-08-04T17:48:25.499182Z","iopub.status.idle":"2023-08-04T17:48:25.680617Z","shell.execute_reply.started":"2023-08-04T17:48:25.499133Z","shell.execute_reply":"2023-08-04T17:48:25.679414Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Definir las columnas para user (customer_id) e item (product_id))\nuser_col = \"customer_id\"\nitem_col = \"product_id\"\n\n# Generar los perfiles de clientes\ncustomer_profiles_df = generate_user_profiles(spark, train_data_ready_for_model_repartitioned, user_col, item_col)\ncustomer_profiles_df.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:48:25.681990Z","iopub.execute_input":"2023-08-04T17:48:25.685013Z","iopub.status.idle":"2023-08-04T17:48:25.857797Z","shell.execute_reply.started":"2023-08-04T17:48:25.684949Z","shell.execute_reply":"2023-08-04T17:48:25.856477Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"customer_profiles_df\")\ncustomer_profiles_df.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:48:25.859237Z","iopub.execute_input":"2023-08-04T17:48:25.861226Z","iopub.status.idle":"2023-08-04T17:48:25.871795Z","shell.execute_reply.started":"2023-08-04T17:48:25.861159Z","shell.execute_reply":"2023-08-04T17:48:25.870353Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"customer_profiles_df.count()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:48:25.874854Z","iopub.execute_input":"2023-08-04T17:48:25.876388Z","iopub.status.idle":"2023-08-04T17:49:01.127317Z","shell.execute_reply.started":"2023-08-04T17:48:25.876332Z","shell.execute_reply":"2023-08-04T17:49:01.125970Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"customer_profiles_df.show(3, truncate=75)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:49:01.130206Z","iopub.execute_input":"2023-08-04T17:49:01.131343Z","iopub.status.idle":"2023-08-04T17:49:01.313225Z","shell.execute_reply.started":"2023-08-04T17:49:01.131295Z","shell.execute_reply":"2023-08-04T17:49:01.311850Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"array_length_udf = udf(calculate_array_length, IntegerType())# Calcular shape (length) de user_interactions array\ncustomer_profiles_df_with_shape = customer_profiles_df.withColumn(\"user_interactions_shape\", array_length_udf(\"user_interactions\"))\ncustomer_profiles_df_with_shape.select(\"customer_id\", \"user_interactions_shape\").show()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:49:01.314957Z","iopub.execute_input":"2023-08-04T17:49:01.317498Z","iopub.status.idle":"2023-08-04T17:49:03.352760Z","shell.execute_reply.started":"2023-08-04T17:49:01.317441Z","shell.execute_reply":"2023-08-04T17:49:03.351900Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Generar product similarities (similaridades entre los productos)\nproduct_profiles_df_repartitioned = product_profiles_df.repartition(100)\nproduct_profiles_df_repartitioned.cache()\nproduct_similarities_df = generate_product_similarities(spark, product_profiles_df_repartitioned)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:49:03.354087Z","iopub.execute_input":"2023-08-04T17:49:03.354544Z","iopub.status.idle":"2023-08-04T17:49:09.955792Z","shell.execute_reply.started":"2023-08-04T17:49:03.354509Z","shell.execute_reply":"2023-08-04T17:49:09.954504Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"product_similarities_df\")\nproduct_similarities_df.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:49:10.240394Z","iopub.execute_input":"2023-08-04T17:49:10.241197Z","iopub.status.idle":"2023-08-04T17:49:10.252498Z","shell.execute_reply.started":"2023-08-04T17:49:10.241153Z","shell.execute_reply":"2023-08-04T17:49:10.251147Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Drop the item_profile field from datasetA and datasetB\nproduct_similarities_df = (\n    product_similarities_df\n        .withColumn(\"datasetA\", F.struct(col(\"datasetA.product_id\")))\n        .withColumn(\"datasetB\", F.struct(col(\"datasetB.product_id\")))\n        .withColumn(\"similarity\", F.col(\"similarity\"))\n)\nproduct_similarities_df.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:49:10.254955Z","iopub.execute_input":"2023-08-04T17:49:10.256936Z","iopub.status.idle":"2023-08-04T17:49:10.342131Z","shell.execute_reply.started":"2023-08-04T17:49:10.256870Z","shell.execute_reply":"2023-08-04T17:49:10.341181Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_similarities_df.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:49:10.343370Z","iopub.execute_input":"2023-08-04T17:49:10.343814Z","iopub.status.idle":"2023-08-04T17:49:10.352301Z","shell.execute_reply.started":"2023-08-04T17:49:10.343772Z","shell.execute_reply":"2023-08-04T17:49:10.351153Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_similarities_df.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:49:10.354101Z","iopub.execute_input":"2023-08-04T17:49:10.355005Z","iopub.status.idle":"2023-08-04T17:49:10.404053Z","shell.execute_reply.started":"2023-08-04T17:49:10.354961Z","shell.execute_reply":"2023-08-04T17:49:10.402885Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_similarities_df.repartition(100).count()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:49:10.406455Z","iopub.execute_input":"2023-08-04T17:49:10.406903Z","iopub.status.idle":"2023-08-04T17:53:06.962651Z","shell.execute_reply.started":"2023-08-04T17:49:10.406862Z","shell.execute_reply":"2023-08-04T17:53:06.960844Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"product_similarities_df.repartition(100).show()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:06.964363Z","iopub.status.idle":"2023-08-04T17:53:06.965312Z","shell.execute_reply.started":"2023-08-04T17:53:06.964964Z","shell.execute_reply":"2023-08-04T17:53:06.964995Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Perform manual stratified split on train_data_ready_for_model based on 'customer_id'\ntraining_data, validation_data = manual_stratified_split(spark, train_data_ready_for_model_repartitioned, 'customer_id', train_ratio=0.8, seed=42)\n\nprint(\"training_data\")\ntraining_data.printSchema()\n\nprint(\"\\nvalidation_data\")\nvalidation_data.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:17.800369Z","iopub.execute_input":"2023-08-04T17:53:17.800808Z","iopub.status.idle":"2023-08-04T17:53:18.131760Z","shell.execute_reply.started":"2023-08-04T17:53:17.800773Z","shell.execute_reply":"2023-08-04T17:53:18.129962Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"training_data.cache()\nvalidation_data.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:22.033480Z","iopub.execute_input":"2023-08-04T17:53:22.037502Z","iopub.status.idle":"2023-08-04T17:53:22.198277Z","shell.execute_reply.started":"2023-08-04T17:53:22.037433Z","shell.execute_reply":"2023-08-04T17:53:22.196818Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"Number of distinct customer_ids in training_data:\", training_data.select(\"customer_id\").distinct().count(), \" | rows: \", training_data.count())","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:06.972149Z","iopub.status.idle":"2023-08-04T17:53:06.972973Z","shell.execute_reply.started":"2023-08-04T17:53:06.972690Z","shell.execute_reply":"2023-08-04T17:53:06.972718Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"check_nulls_in_dataframe(training_data)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:27.598144Z","iopub.execute_input":"2023-08-04T17:53:27.598608Z","iopub.status.idle":"2023-08-04T17:53:39.385361Z","shell.execute_reply.started":"2023-08-04T17:53:27.598564Z","shell.execute_reply":"2023-08-04T17:53:39.383166Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"check_nulls_in_dataframe(validation_data)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:06.977181Z","iopub.status.idle":"2023-08-04T17:53:06.978097Z","shell.execute_reply.started":"2023-08-04T17:53:06.977771Z","shell.execute_reply":"2023-08-04T17:53:06.977801Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Preparar el modelo ALS\nals = build_model_ALS(spark, training_data)\n\n# Crear el parameter grid para hyperparameter tuning\nparam_grid = (\n        ParamGridBuilder()\n            .addGrid(als.rank, [5, 10, 15])\n            .addGrid(als.regParam, [0.01, 0.1, 1.0])\n            .build()\n    )\n\n# Definir la métrica de evaluación (p.ej., RMSE, R2) para la selección del modelo\nevaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"combined_rating\", predictionCol=\"prediction\")\n\n# Crear el TrainValidationSplit para hyperparameter tuning\ntvs = TrainValidationSplit(\n        estimator=als,\n        estimatorParamMaps=param_grid,\n        evaluator=evaluator,\n        trainRatio=0.8,  # 80% of the training set will be used for training\n        seed=42\n    )","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:44.619763Z","iopub.execute_input":"2023-08-04T17:53:44.620331Z","iopub.status.idle":"2023-08-04T17:53:44.680375Z","shell.execute_reply.started":"2023-08-04T17:53:44.620281Z","shell.execute_reply":"2023-08-04T17:53:44.678980Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"training_data_repartitioned = training_data.repartition(100)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:52.391189Z","iopub.execute_input":"2023-08-04T17:53:52.391793Z","iopub.status.idle":"2023-08-04T17:53:52.407759Z","shell.execute_reply.started":"2023-08-04T17:53:52.391744Z","shell.execute_reply":"2023-08-04T17:53:52.406009Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"training_data_repartitioned.cache()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T18:00:16.125163Z","iopub.execute_input":"2023-08-04T18:00:16.126272Z","iopub.status.idle":"2023-08-04T18:00:16.168763Z","shell.execute_reply.started":"2023-08-04T18:00:16.126214Z","shell.execute_reply":"2023-08-04T18:00:16.167327Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Fit TrainValidationSplit para encontrar el mejor modelo\nmodel = tvs.fit(training_data_repartitioned)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T18:00:23.600664Z","iopub.execute_input":"2023-08-04T18:00:23.601296Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Hacer predicciones con el set de validación\npredictions = model.transform(validation_data)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:06.989978Z","iopub.status.idle":"2023-08-04T17:53:06.991103Z","shell.execute_reply.started":"2023-08-04T17:53:06.990774Z","shell.execute_reply":"2023-08-04T17:53:06.990807Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Evaluar la performance del modelo en el set de validación\nrmse = evaluator.evaluate(predictions)\n\n# Print RMSE del set de validación\nprint(\"Root Mean Squared Error (RMSE) en set de validación = \" + str(rmse))","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:06.993028Z","iopub.status.idle":"2023-08-04T17:53:06.994030Z","shell.execute_reply.started":"2023-08-04T17:53:06.993741Z","shell.execute_reply":"2023-08-04T17:53:06.993769Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Obtener el mejor modelo del TrainValidationSplit\nbest_model = model.bestModel\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:06.995859Z","iopub.status.idle":"2023-08-04T17:53:06.996816Z","shell.execute_reply.started":"2023-08-04T17:53:06.996508Z","shell.execute_reply":"2023-08-04T17:53:06.996537Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Generar dataframe con 30 SKUs recomendados por customer_id\nsubmission_df = generar_30_SKUs_recomendados(spark, train_data_ready_for_model_repartitioned, best_model)\nprint(\"submission_df\")\nprint(submission_df.count())\nsubmission_df.printSchema()\nsubmission_df.show(5, truncate=150)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:06.998691Z","iopub.status.idle":"2023-08-04T17:53:06.999689Z","shell.execute_reply.started":"2023-08-04T17:53:06.999378Z","shell.execute_reply":"2023-08-04T17:53:06.999408Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"customer_purchases_df = generate_customer_purchases(spark, train_data)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.002793Z","iopub.status.idle":"2023-08-04T17:53:07.004785Z","shell.execute_reply.started":"2023-08-04T17:53:07.003929Z","shell.execute_reply":"2023-08-04T17:53:07.003967Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Para crear recomendaciones híbridas que tengan en cuenta tanto las similitudes de los elementos ALS \n# como las basadas en el contenido, actualicemos el planteamiento del siguiente modo:\n\n# Para clasificar las recomendaciones basadas en el contenido para cada usuario, tendrá que filtrar el DataFrame \n# product_similarities_df basándose en los productos con los que el usuario ha interactuado (comprado). \n# Luego ordenar los elementos recomendados por ranking de similitud en orden descendente.\ngrouped_recommendations_df = generate_grouped_recommendations(spark, customer_purchases_df, product_similarities_df)\n# print(\"grouped_recommendations_df\")\n# grouped_recommendations_df.printSchema()\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.007314Z","iopub.status.idle":"2023-08-04T17:53:07.008450Z","shell.execute_reply.started":"2023-08-04T17:53:07.008128Z","shell.execute_reply":"2023-08-04T17:53:07.008160Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(grouped_recommendations_df.count())","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.010700Z","iopub.status.idle":"2023-08-04T17:53:07.011780Z","shell.execute_reply.started":"2023-08-04T17:53:07.011448Z","shell.execute_reply":"2023-08-04T17:53:07.011481Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"grouped_recommendations_df = grouped_recommendations_df.repartition(10)  # Adjust the number of partitions as needed\ngrouped_recommendations_df.limit(10).show()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.013776Z","iopub.status.idle":"2023-08-04T17:53:07.014854Z","shell.execute_reply.started":"2023-08-04T17:53:07.014513Z","shell.execute_reply":"2023-08-04T17:53:07.014544Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"grouped_recommendations_df.show(5, truncate=150)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.016796Z","iopub.status.idle":"2023-08-04T17:53:07.017818Z","shell.execute_reply.started":"2023-08-04T17:53:07.017500Z","shell.execute_reply":"2023-08-04T17:53:07.017530Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Sabiendo que ya hemos creado las recomendaciones ALS DataFrame \"submission_df\",\n# el DataFrame de recomendaciones basadas en el contenido \"grouped_recommendations_df\", \n# y el DataFrame de similitudes de artículos \"product_similarities_df\".\n\n# Definir los factores de ponderación para las recomendaciones ALS, basadas en el contenido y en las similitudes de los ítems.\nweight_als = 0.5\nweight_content_based = 0.5\nmodel_als_elements_to_select = int(30 * weight_als)\nmodel_content_elements_to_select = 30 - model_als_elements_to_select\n\n# Aplicar ponderaciones a las recomendaciones sobre ALS, basadas en el contenido y en las similitudes de los ítems.\nA = submission_df\nB = grouped_recommendations_df","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.019683Z","iopub.status.idle":"2023-08-04T17:53:07.020694Z","shell.execute_reply.started":"2023-08-04T17:53:07.020380Z","shell.execute_reply":"2023-08-04T17:53:07.020409Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Seleccione los primeros elementos model_als_elements_to_select de product_id_list en dataframe A\nselected_als_elements_a_product = F.slice(A[\"product_id_list\"], 1, model_als_elements_to_select)\nselected_als_elements_a_rank = F.slice(A[\"rank_list\"], 1, model_als_elements_to_select)\n\n# Seleccione los primeros elementos model_content_elements_to_select de product_id_list en dataframe B\nselected_content_elements_b_product = F.slice(B[\"product_id_list\"], 1, model_content_elements_to_select)\nselected_content_elements_b_rank = F.slice(B[\"rank_list\"], 1, model_content_elements_to_select)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.022560Z","iopub.status.idle":"2023-08-04T17:53:07.023564Z","shell.execute_reply.started":"2023-08-04T17:53:07.023263Z","shell.execute_reply":"2023-08-04T17:53:07.023293Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Aplicar Join a ambos dataframes usando customer_id y user_id\nhybrid_submision_df = A.join(B, A[\"customer_id\"] == B[\"user_id\"], \"inner\")","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.025468Z","iopub.status.idle":"2023-08-04T17:53:07.026432Z","shell.execute_reply.started":"2023-08-04T17:53:07.026138Z","shell.execute_reply":"2023-08-04T17:53:07.026168Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Crear las nuevas columnas 'combined_product_id_list' y combined_rank_list\nhybrid_submision_df = (\n        hybrid_submision_df\n            .withColumn(\n                \"combined_product_id_list\",\n                F.concat(selected_als_elements_a_product, selected_content_elements_b_product)\n            )\n            .withColumn(\n                \"combined_rank_list\",\n                F.concat(selected_als_elements_a_rank, selected_content_elements_b_rank)\n            )\n    )\n\nprint(\"hybrid_submision_df\")\nhybrid_submision_df.printSchema()","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.028335Z","iopub.status.idle":"2023-08-04T17:53:07.029326Z","shell.execute_reply.started":"2023-08-04T17:53:07.028990Z","shell.execute_reply":"2023-08-04T17:53:07.029020Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Aplicar la función UDF zip_and_sort para crear el DataFrame final con la estructura deseada\nreformatted_hybrid_submision_df = (\n        hybrid_submision_df\n        .withColumn(\"final_recommendations\", zip_and_sort(\n            F.col(\"combined_product_id_list\"), \n            F.col(\"combined_rank_list\")\n        ))\n        .select(\"customer_id\", \"final_recommendations\")\n        .withColumnRenamed(\"final_recommendations\", \"product_id\")\n    )\nprint(\"reformatted_hybrid_submision_df\")\nreformatted_hybrid_submision_df.printSchema()\n# reformatted_hybrid_submision_df.show(truncate=False)","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.031253Z","iopub.status.idle":"2023-08-04T17:53:07.032244Z","shell.execute_reply.started":"2023-08-04T17:53:07.031919Z","shell.execute_reply":"2023-08-04T17:53:07.031947Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Generar archivo submission###.csv\nsave_recommendations_to_submission(spark, reformatted_hybrid_submision_df, OUTPUT_DIRECTORY)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.034127Z","iopub.status.idle":"2023-08-04T17:53:07.035236Z","shell.execute_reply.started":"2023-08-04T17:53:07.034822Z","shell.execute_reply":"2023-08-04T17:53:07.034850Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Obtener dataframe recall_por_cliente (customer_id, recall)\nrecall_por_cliente = call_funcion_get_recall_por_cliente(spark, DIRECTORY, train_data, reformatted_hybrid_submision_df)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.037015Z","iopub.status.idle":"2023-08-04T17:53:07.037983Z","shell.execute_reply.started":"2023-08-04T17:53:07.037689Z","shell.execute_reply":"2023-08-04T17:53:07.037719Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Calcular la media de recall\naverage_recall = recall_por_cliente.agg(F.mean(\"recall\")).collect()[0][0]\nprint(\"El promedio de Recall es: \", average_recall)\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.039792Z","iopub.status.idle":"2023-08-04T17:53:07.040760Z","shell.execute_reply.started":"2023-08-04T17:53:07.040457Z","shell.execute_reply":"2023-08-04T17:53:07.040485Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"spark.stop\n","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.042541Z","iopub.status.idle":"2023-08-04T17:53:07.043506Z","shell.execute_reply.started":"2023-08-04T17:53:07.043220Z","shell.execute_reply":"2023-08-04T17:53:07.043249Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!pip install numpy==1.22.2","metadata":{"execution":{"iopub.status.busy":"2023-08-04T17:53:07.045312Z","iopub.status.idle":"2023-08-04T17:53:07.046243Z","shell.execute_reply.started":"2023-08-04T17:53:07.045912Z","shell.execute_reply":"2023-08-04T17:53:07.045943Z"},"trusted":true},"execution_count":null,"outputs":[]}]}