1

Estoy intentando realizar un análisis de correlación de variables. Para lo cual tengo la información en un RDD, lo convierto a DataFrame con Pandas. Me aseguro que todos los datos sean numéricos y finalmente realizo una matriz con su respectivo mapa de calor para encontrar su correlación. El código es el siguiente:

import matplotlib.pyplot as plt
import pandas as pd 
import seaborn as sns

ctes = spark.read.parquet(path)
####################################################################################
################ Correlación de Variables Numéricas ################################
####################################################################################

# Lista de Variables Numéricas
var_num = ['n_reclamaciones', 'monto_total', 'n_tarjeta', 'fechas_op_dist','fechas_ini_dist','dist_fech_max',
'monto_promedio', 'dif_monto_max_min', 'dictamen_afavor_total', 'NO_CLAIMS_1_MONTH'
]

#Convertir a DF
tabal_aux = ctes[var_num].toPandas()

# ------------------Correlacion Spearman----------------------
tabla_corr = tabal_aux
#Cliclo para asegurar que todo tipo de datos sea numérico
for var in var_num:
    tabla_corr[var] = tabla_corr[var].astype(float)
corr = tabla_corr.corr(method = 'spearman')
plt.figure(figsize = (20,10))
sns.heatmap(corr, xticklabels = corr.columns.values, yticklabels = corr.columns.values, annot = True, linewidths = .6, cmap = 'Blues')
plt.tittle('Tabla de correlacioón Spearman')
plt.show()
plt.clf()

# ------------------Correlacion Pearson----------------------
tabla_corr = tabal_aux
for var in var_num:
    tabla_corr[var] = tabla_corr[var].astype(float)
corr = tabla_corr.corr(method = 'pearson')
plt.figure(figsize = (20,10))
sns.heatmap(corr, xticklabels = corr.columns.values, yticklabels = corr.columns.values, annot = True, linewidths = .6, cmap = 'Blues')
plt.tittle('Tabla de correlacioón Pearson')
plt.show()
plt.clf()

Sin embargo al ejecutar me aparece el siguiente error por consola (adjunto imagen):

introducir la descripción de la imagen aquí

Les dejo la traza del error completa en formato código a continuación:

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<ipython-input-8-38257abcbb84> in <module>()
     20            'MAX_LIMITE_CREDITO', 'MAX_PORC_ENDEUDAMIENTO', 'MIN_PORC_ENDEUDAMIENTO']
     21 # Convertir a df
---> 22 tabla_aux =  ctes[var_num].toPandas()
     23 # ----------------------------------------------- Correlación de Spearman -----------------------------------------------
     24 tabla_corr = tabla_aux
/opt/spark/dist/python/pyspark/sql/dataframe.py in toPandas(self)
   1701         """
   1702         import pandas as pd
-> 1703         return pd.DataFrame.from_records(self.collect(), columns=self.columns)
   1704 
   1705     ##########################################################################################
/opt/spark/dist/python/pyspark/sql/dataframe.py in collect(self)
    436         """
    437         with SCCallSiteSync(self._sc) as css:
--> 438             port = self._jdf.collectToPython()
    439         return list(_load_from_socket(port, BatchedSerializer(PickleSerializer())))
    440 
/opt/spark/dist/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1131         answer = self.gateway_client.send_command(command)
   1132         return_value = get_return_value(
-> 1133             answer, self.gateway_client, self.target_id, self.name)
   1134 
   1135         for temp_arg in temp_args:
/opt/spark/dist/python/pyspark/sql/utils.py in deco(*a, **kw)
     61     def deco(*a, **kw):
     62         try:
---> 63             return f(*a, **kw)
     64         except py4j.protocol.Py4JJavaError as e:
     65             s = e.java_exception.toString()
/opt/spark/dist/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    317                 raise Py4JJavaError(
    318                     "An error occurred while calling {0}{1}{2}.\n".
--> 319                     format(target_id, ".", name), value)
    320             else:
    321                 raise Py4JError(
Py4JJavaError: An error occurred while calling o319.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 17 tasks (1060.8 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
    at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1516)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1504)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1503)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
    at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1503)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:814)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:814)
    at scala.Option.foreach(Option.scala:257)
    at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:814)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1731)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1686)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1675)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
    at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:630)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2051)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2072)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2091)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2116)
    at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:936)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
    at org.apache.spark.rdd.RDD.collect(RDD.scala:935)
    at org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:278)
    at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply$mcI$sp(Dataset.scala:2804)
    at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:2801)
    at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:2801)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:65)
    at org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:2824)
    at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:2801)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:280)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:214)
    at java.lang.Thread.run(Thread.java:748)

Entiendo por el error que me arroja que la conversión de RDD a DF no le gusta, o al menos es lo que interpreto. Podrían ayudarme orientándome en cual es mi problema en el código. De antemano gracias.

2
  • Por favor, podrías agregar la traza del error completa y en formato código, así será más fácil ayudarte. el 19 oct. 2020 a las 7:34
  • Listo, ya he añadido la traza completa del error. el 19 oct. 2020 a las 14:08

1 respuesta 1

0

Ya pude lograr lo que necesitaba. Ósea la matriz de correlación. Añado el código con el que lo conseguí por si alguien se enfrenta en un futuro con un problema similar:

from pyspark.mllib.stat import Statistics
# Lista de variables numéricas
var_num = ['n_reclamaciones', 'monto_total', 'n_tarjeta', 'n_com', 'fechas_op_dist', 'fechas_inc_dist',
           'dist_fechas_max', 'monto_promedio', 'dif_monto_max_min', 'dictamen_afavor_total', 'NO_CLAIMS_1_MONTH',
          'NO_CLAIMS_3_MONTH', 'NO_CLAIMS_6_MONTH', 'NO_CLAIMS_12_MONTH', 'int_dias_prom', 'monto_1_mes',  'monto_3_mes',
          'monto_6_mes', 'monto_12_mes', 'num_fechas_aclaracion_dist', 'avg_claims_per_week','dif_monto_first_last',
          'div_comercio', 'edad', 'antiguedad',
          'pendiente', 'DISTINTAS_TARJETAS', 'DISTINTOS_GIROS', 'DISTINTAS_AFILIACIONES', 'NUMERO_TRANSACCIONES',
           'NUMERO_COMPRAS', 'NUMERO_RETIROS', 'MONTO_TRANSACCIONES', 'MONTO_COMPRAS', 'MONTO_RETIROS',
           'COMPRAS_CREDITO', 'COMPRAS_DEBITO', 'RETIROS_CREDITO', 'RETIROS_DEBITO', 'COMPRAS_MADRUGADA',
           'COMPRAS_MANANA', 'COMPRAS_TARDE', 'COMPRAS_NOCHE', 'RETIROS_MADRUGADA', 'RETIROS_MANANA',
           'RETIROS_TARDE', 'RETIROS_NOCHE', 'COMPRAS_ENTRE_SEMANA', 'COMPRAS_FIN_DE_SEMANA', 'RETIROS_ENTRE_SEMANA',
           'RETIROS_FIN_DE_SEMANA', 'COMPRAS_SUPERMERCADO', 'COMPRAS_AUTOS', 'COMPRAS_ENTRETENIMIENTO',
           'COMPRAS_VIAJES', 'COMPRAS_COMIDA', 'COMPRAS_ECOMMERCE', 'COMPRAS_TRANSPORTE', 'COMPRAS_ROPA_Y_CALZADO',
           'COMPRAS_EDUCACION', 'COMPRAS_DEPORTES', 'COMPRAS_TIENDAS_DEP', 'COMPRAS_SALUD', 'COMPRAS_CONSTRUCCION',
           'COMPRAS_PAGO_SERVICIOS', 'COMPRAS_MUEBLES_Y_OFICINAS', 'COMPRAS_BENEFICENCIA', 'COMPRAS_HOGAR',
           'COMPRAS_OTROS', 'MAX_CREDITO_DISPONIBLE', 'MAX_DEBITO_DISPONIBLE', 'MAX_SALDO_CREDITO',
           'MAX_LIMITE_CREDITO', 'MAX_PORC_ENDEUDAMIENTO', 'MIN_PORC_ENDEUDAMIENTO']
# Convertir a df
df =  ctes[var_num] 
def compute_correlation_matrix(df, method='pearson'):
    df_rdd = df.rdd.map(lambda row: row[0:])
    corr_mat = Statistics.corr(df_rdd, method=method)
    corr_mat_df = pd.DataFrame(corr_mat,
                    columns=df.columns,
                    index=df.columns)
    return corr_mat_df
compute_correlation_matrix(df)

Como salida me da la matriz de correlación de Pearson: introducir la descripción de la imagen aquí

Tu Respuesta

By clicking “Publica tu respuesta”, you agree to our terms of service and acknowledge you have read our privacy policy.

¿No es la respuesta que buscas? Examina otras preguntas con la etiqueta o formula tu propia pregunta.