PySpark funciones
En este artículo mostraré algunas funciones de PySpark recomendadas que debes saber antes de mi siguiente articulo sobre análisis para clientes.
¿Qué es Pyspark?
Spark se ha incorporado herramientas de la mayoría de los científicos de datos. Es un framework open source para la computación en paralelo utilizando clusters. Se utiliza especialmente para acelerar la computación iterativa de grandes cantidades de datos o de modelos muy complejos.
- Python -> PySpark
Acontinuación mostrare los métodos tal vez mas usado por la implicancia de los datos o tal vez menos usado por unos o posiblemente no lo hayas tomado en cuenta =)
METODOS
Lit
df.withColumn('new', F.lit('A')) 
Coalesce
df = spark.createDataframe([(None,None),(1,None),(None,2)],(a,b)) 
df.select(coalesce(df['a'], df['b']))
Collect_set y Collect_list
df = sqlContext.createDataFrame([('a', None, None),('a', 'code1', None) ('a', 'code2', 'name2'),], ['id', 'code', 'name']) 
df2 = df.groupby('id').agg(collect_set('code'), collect_list('name')) 
Concat
df = sqlContext.createDataFrame([('a', None, None),('a', 'code1', None),
('a', 'code2', 'name2'),], ['id', 'code', 'name']) 
df2 = df.select(concat(df['code'], df['name']).alias('concadenado')) 
Bin
df = sqlContext.createDataFrame([('2'),('3')], ['numero']) 
df2 = df.select(bin(df['numero']).alias('binario')) 
Explode
df = spark.createDataFrame([(1, 'A', [1,2,3]), (2, 'B', [3,5])],['col1', 'col2', 'col3']) 
df2 = df.withColumn('col3', explode(df.col3)) 
Window
df = spark.createDataFrame([{'id': 123, 'dt': 0}, {'id': 456, 'dt':0}, {'id':456, 'dt':2}]) 
w = Window.partitionBy('id').orderBy('dt')
df2 = df.select('id','dt', count('dt').over(w).alias('count')) 
Udf
df = spark.createDataFrame([{'id': 1, 'name': 'jonathan'}, {'id': 2, 'name':'jonathan1'}, {'id':3, 'name':'jonathan2'}]) 
df2 = df.withColumn('category', udfScoreToCategory(df['score'])) 
¿Porque aprender Metodos?
Aprender un poco de los métodos, aveces necesitamos traducir queries enormes para trabajar para crear modelos de Machine Learning y como serán funciones que se ejecuten en tiempo real o según el uso, debes tener algo armado para no estar repitiendo siempre lo mismo.
Ejemplo simple traducción de un queries complejo y eso que aveces tenemos queries de esta magnitud para consultar para fuentes de información, nosotros como científicos de datos tenemos que entender para poder dar soluciones y que el modelo responda a las preguntas del negocio.
Usando sparkSql
traducción en código una parte en este caso “colita de chancho salvo el código” algunas veces las funciones nos salvan ~ (ISIN)
df_group = df.groupby(“passengerID”)\
.agg(F.sum(F.when(F.col(“hour”).isin(types_manana1),1).otherwise(0)).alias(“manana1”),
F.sum(F.when(F.col(“hour”).isin(types_manana2),1).otherwise(0)).alias(“manana2”),
F.sum(F.when(F.col(“hour”).isin(types_tarde),1).otherwise(0)).alias(“tarde”),
F.sum(F.when(F.col(“hour”).isin(types_noche1),1).otherwise(0)).alias(“noche1”),
F.sum(F.when(F.col(“hour”).isin(types_noche2),1).otherwise(0)).alias(“noche2”),
F.sum(F.when(F.col(“hour”).isin(types_juerga),1).otherwise(0)).alias(“juerga”),
F.sum(F.when(~F.col(“weekday”).isin(type_weekday),1).otherwise(0)).alias(“semana”),
F.sum(F.when(F.col(“weekday”).isin(type_weekday),1).otherwise(0)).alias(“fin_de_semana”),
F.sum(“payment_value”).alias(“monto_total”),
F.avg(“taximeter_distance”).alias(“distance”),
F.round(F.datediff(F.max(“requested_at”), F.min(“requested_at”))/F.count(“*”),2).alias(“tiempo_entre_viajes”)
)
df_group2 = df_group.withColumn(“v_total”, F.col(“manana1”)+ F.col(“manana2”)+ F.col(“tarde”)+ F.col(“noche1”)+ F.col(“noche2”)+ F.col(“juerga”))
df_group2 = df_group2.withColumn(“v_total_semana”, F.col(“semana”)+ F.col(“fin_de_semana”))
df_group3 = df_group2.withColumn(“v_manana1”, df_group2[“manana1”])
df_group3 = df_group3.withColumn(“v_manana2”, df_group2[“manana2”])
df_group3 = df_group3.withColumn(“v_tarde”, df_group2[“tarde”])
df_group3 = df_group3.withColumn(“v_noche1”, df_group2[“noche1”])
df_group3 = df_group3.withColumn(“v_noche2”, df_group2[“noche2”])
df_group3 = df_group3.withColumn(“v_juerga”, df_group2[“juerga”])
df_group3 = df_group3.withColumn(“v_semana”, df_group2[“semana”])
df_group3 = df_group3.withColumn(“v_fin_de_semana”, df_group2[“fin_de_semana”])
df_group3 = df_group3.withColumn(“v_avg_distance”, F.round(df_group2[“distance”],2))
df_group3 = df_group3.withColumn(“v_avg_payment”, F.round(df_group2[“monto_total”],2)) continua…..
Espero le hayas gustado esta publicación compartiendo experiencia en los que vienen hablaremos un poco de análisis de Cohort , CLV, Survived, RFM , Métodos supervisados o no supervisado en pyspark para su segmentación o analítica de clientes para observar su comportamiento de uso.
linkedin:
