Skip to content

magdielgutierrez/Analisis-de-datos-de-Amazon-usando-Apache-Spark-PySpark-

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

45 Commits
 
 
 
 

Repository files navigation

Analisis de datos de Amazon usando Apache Spark (PySpark)

Realizar un flujo ETL usando herramientas como Airbyte para extraer los datos, almacenarlo en un capa de stagging en Google Cloud Storage y Bigquery para las transformaciones y acciones basadas en la API de Spark en Dataproc que van desde simples hasta bastante complejas. Se mantienen dos fuentes de datos:

1-Estática: Base de datos en Cockroach con las siguiente tablas: clients, tasas_cambio_pais_anual,compras, external_products_products

1.1 Estática: Bucket de GCS llamado zophia-proyecto-final-de tabla de historico de compras.

2-Dinámica: dataset amazon_daily_updates en Bigquery que diariamente recibe nuevos registros de las compras realizadas en el día, que se agregan a la tabla compras.

Flujo completo:

Contenido

Una breve sinopsis de lo que es cada caso de uso y qué funcionalidad de SPARK SQL se uso.

Sección Funciones
1. Revisando el Data Set Cockroach
2. Extracción de la data de Cockroach a una capa de staging GCS
3. Extracción de la data GCS a una capa de staging BigQuery
4. Transformación y limpieza de la data
4.1 Creando tabla de productos REGEXP_EXTRACT, REGEXP_REPLACE, WHEN, TRANSLATE, COL, GROUP BY, AGG, ORDERBY, CONCAT, LAST, INNER JOIN
4.2 Creando tabla promedio de precio de productos COUNTDISTINCT, MEAN, GROUP BY, AGG, SORT
4.3 Creando tabla rango de precios de productos GROUP BY, AGG, FIRST, LAST, MIN, MAX
4.4 Creando tabla promedio de evaluación COUNTDISTINCT, AVG, GROUP BY, AGG
4.5 Creando tabla de compras TODATE, COL
4.6 Creando tabla de compras anuales COUNTDISTINCT, MONTH, COUNT, SUM, AVG, WINDOWS, LAG
4.7 Creando tabla de compras mensuales YEAR, COUNTDISTINCT, COUNT, SUM, AVG, COL, INNER JOIN
5. Tabla de hechos YEAR, SUM, COUNTDISTINCT, GROUPBY, AGG, SORT, COL , INNER JOIN
6. Información de cargas incrementales desde BigQuery
6.1 Carga de datos de compras diarias FILTER, CAST
6.2 Calculo de compras anuales FILTER, DATE_TRUNC, ADD_MONTHS, COL, MONTH, YEAR, GROUPBY, AGG,COUNT, COUNTDISTINCT, SUM, AVG, INNER JOIN, SORT
6.3 Calculo de compras mensuales FILTER, DATE_TRUNC, ADD_MONTHS, COL, MONTH, YEAR, GROUPBY, AGG, COUNT, COUNTDISTINCT, SUM, AVG, INNER JOIN

1. Revisando el Data Set Cockroach

Tablas de data set de Amazon:

image

Descripcion de tablas:

>SHOW COLUMNS FROM clients

column_name   |data_type   |is_nullable|column_default|generation_expression|indices  |is_hidden|
--------------+------------+-----------+--------------+---------------------+---------+---------+
id            |VARCHAR(20) |true       |              |                     |{primary}|false    |
nombre        |VARCHAR(150)|true       |              |                     |{primary}|false    |
direccion     |VARCHAR(200)|true       |              |                     |{primary}|false    |
email         |VARCHAR(100)|true       |              |                     |{primary}|false    |
telefono      |VARCHAR(13) |true       |              |                     |{primary}|false    |
numero_tarjeta|VARCHAR(20) |true       |              |                     |{primary}|false    |
isprime       |VARCHAR     |true       |              |                     |{primary}|false    |
rowid         |INT8        |false      |unique_rowid()|                     |{primary}|true     |

>SHOW COLUMNS FROM compras

column_name |data_type|is_nullable|column_default|generation_expression|indices  |is_hidden|
------------+---------+-----------+--------------+---------------------+---------+---------+
id          |STRING   |true       |              |                     |{primary}|false    |
client_id   |STRING   |true       |              |                     |{primary}|false    |
product_id  |STRING   |true       |              |                     |{primary}|false    |
cantidad    |INT8     |true       |              |                     |{primary}|false    |
precio      |FLOAT8   |true       |              |                     |{primary}|false    |
envio_id    |STRING   |true       |              |                     |{primary}|false    |
isprime     |STRING   |true       |              |                     |{primary}|false    |
fecha_compra|DATE     |true       |              |                     |{primary}|false    |
metodo_pago |STRING   |true       |              |                     |{primary}|false    |
rowid       |INT8     |false      |unique_rowid()|                     |{primary}|true     |

>SHOW COLUMNS FROM products

column_name            |data_type   |is_nullable|column_default|generation_expression|indices  |is_hidden|
-----------------------+------------+-----------+--------------+---------------------+---------+---------+
isbestseller           |VARCHAR     |true       |              |                     |{primary}|false    |
product_title          |VARCHAR(500)|true       |              |                     |{primary}|false    |
product_main_image_url |VARCHAR(100)|true       |              |                     |{primary}|false    |
app_sale_price         |VARCHAR(10) |true       |              |                     |{primary}|false    |
app_sale_price_currency|VARCHAR(5)  |true       |              |                     |{primary}|false    |
isprime                |VARCHAR     |true       |              |                     |{primary}|false    |
product_detail_url     |VARCHAR(50) |true       |              |                     |{primary}|false    |
product_id             |VARCHAR(20) |true       |              |                     |{primary}|false    |
evaluate_rate          |VARCHAR(50) |true       |              |                     |{primary}|false    |
original_price         |VARCHAR(15) |true       |              |                     |{primary}|false    |
country                |VARCHAR(2)  |true       |              |                     |{primary}|false    |
rowid                  |INT8        |false      |unique_rowid()|                     |{primary}|true     |

>SHOW COLUMNS FROM external_products

column_name            |data_type   |is_nullable|column_default|generation_expression|indices  |is_hidden|
-----------------------+------------+-----------+--------------+---------------------+---------+---------+
isbestseller           |VARCHAR     |true       |              |                     |{primary}|false    |
product_title          |VARCHAR(500)|true       |              |                     |{primary}|false    |
product_main_image_url |VARCHAR(250)|true       |              |                     |{primary}|false    |
app_sale_price         |VARCHAR(15) |true       |              |                     |{primary}|false    |
app_sale_price_currency|VARCHAR(3)  |true       |              |                     |{primary}|false    |
isprime                |VARCHAR     |true       |              |                     |{primary}|false    |
product_detail_url     |VARCHAR(250)|true       |              |                     |{primary}|false    |
product_id             |VARCHAR(15) |true       |              |                     |{primary}|false    |
evaluate_rate          |VARCHAR(50) |true       |              |                     |{primary}|false    |
original_price         |VARCHAR(15) |true       |              |                     |{primary}|false    |
country                |VARCHAR(2)  |true       |              |                     |{primary}|false    |
rowid                  |INT8        |false      |unique_rowid()|                     |{primary}|true     |


>SHOW COLUMNS FROM tasas_cambio_pais_anual

column_name |data_type  |is_nullable|column_default|generation_expression|indices  |is_hidden|
------------+-----------+-----------+--------------+---------------------+---------+---------+
Country-name|VARCHAR(50)|true       |              |                     |{primary}|false    |
Alpha-2-code|VARCHAR(2) |true       |              |                     |{primary}|false    |
Alpha-3-code|VARCHAR(3) |true       |              |                     |{primary}|false    |
currency    |VARCHAR(5) |true       |              |                     |{primary}|false    |
Year        |INT8       |true       |              |                     |{primary}|false    |
value       |FLOAT4     |true       |              |                     |{primary}|false    |
rowid       |INT8       |false      |unique_rowid()|                     |{primary}|true     |

Back to Top

2. Extracción de la data de Cockroach a una capa de staging Google Cloud Storage

Actividades:

  1. Crear una fuente en Airbyte.

  2. Crear un destino en Airbyte con el conector sea de tipo Google Cloud Storage (GCS), con output format CSV .

  3. Crear una conexión entre la fuente y el destino para extraer las tablas.

Conexión fuente-destino

Archivos CSV en GCS

Back to Top

3. Extracción de la data GCS a una capa de staging BigQuery

Se realiza la extracción de los archivos CSV bucket amazon_magdielgutierrez y zophia-proyecto-final-de a dataset en becade_mgutierrez en Bigquery

Actividades:

  1. Crear un dataset nuevo en BigQuery
  2. Crear tablas de con el prefijo stg_{table}

Actividades realizadas desde la consola de Google Platform

Captura de pantalla: resultado de nuevo dataset en Bigquery

02

Back to Top

4. Transformación y limpieza de la data

Retos de la data a tranformar

  • Formato de fecha TIMESTAMP a DATE
  • Datos tipo BOLEANN a STRING
  • Datos tipo LONG a INT
  • Uso de expresiones regulares para extraer evaluacion de productos dato tipo STRING a DOUBLE
  • Uso de expresiones regulares para eliminar carácteres [coma, puntos, simbolos] de precio de productos dato tipo STRING a DOUBLE
  • Renombrar columnas y nombre de datos en diferentes tablas para cumplir los estándares del naming convention

4.1 Creando tabla de productos

La tabla pr_products es creada a partir de stg_products y stg_external_productos

Cargando datos a dataframe:

#name table products
table_products = "becade_mgutierrez.stg_products" # or becade_mgutierrez.stg_external_productos

#load table to dataframe
stg_products = spark.read \
  .format("bigquery") \
  .option("table", table_products) \
  .load()
   
#select columns from table
raw_products = stg_products.select('product_id','country','app_sale_price','evaluate_rate','isbestseller','isprime','app_sale_price_currency')

#clean column app_sale_price drop values 'None'
raw_products = raw_products.where(raw_products.app_sale_price != 'None')

# fill empty rows evaluate_rate
df_raw_products= raw_products.withColumn("evaluate_rate", when(col("evaluate_rate")=="" ,None)  \
                               .otherwise(col("evaluate_rate"))) 
                               
#clean column app_sale_price drop values 'None'
df_raw_products = df_raw_products.where(df_raw_products.evaluate_rate != "None")

#drop duplicates rows products
df_raw_products = df_raw_products.dropDuplicates()


#clean column evaluate_rate extract format {n.n} &&  replace characters {,} by {.}
df_clean_rate = df_raw_products \
                .withColumn('clean_rate', regexp_extract(col('evaluate_rate'), r'([0-9][\.\,][0-9])',1)) \
                .withColumn('clean_rate', translate(col('clean_rate'), ',', '.'))
                
#concat columns  number_price + decimal_price = app_sale_price_us
df_clean_products_raw=df_raw_price.select('product_id','country','isbestseller','isprime','app_sale_price_currency','clean_rate',
                                          concat(df_raw_price.number_price,df_raw_price.decimal_price).alias("app_sale_price"))
                                          

Mostrando los resultados de tranformación:

ANTES:

B081VKWHY2|JP     |true        |false  |             |¥1,547       |5つ星のうち4.4       |
B07Q2HL42Y|JP     |true        |false  |             |¥2,273       |5つ星のうち4.6       |
B07N4B5MBK|BR     |true        |false  |R$           |2.000,35      |4,5 de 5 estrelas |
B00LI4O9EA|BR     |false       |false  |R$           |275,00        |4,8 de 5 estrelas |

DESPUES:

+----------+-------+------------+-------+-----------------------+----------+--------------+
|product_id|country|isbestseller|isprime|app_sale_price_currency|clean_rate|app_sale_price|
+----------+-------+------------+-------+-----------------------+----------+--------------+
|B081VKWHY2|     JP|        true|  false|                       |       4.4|          1547|
|B07Q2HL42Y|     JP|        true|  false|                       |       4.6|          2273|
|B07N4B5MBK|     BR|        true|  false|                    R$ |       4.5|       2000.35|
|B00LI4O9EA|     BR|       false|  false|                    R$ |       4.8|           275|
+----------+-------+------------+-------+-----------------------+----------+--------------+

Los precios actuales [app_sale_price] estan en la moneda local de cada pais, necesitos convertirlo en dolar, para ello tenemos la tabla pr_tasas_cambio_pais_anual

Cargando datos a dataframe:

#name table exchange
table_exchange = "becade_mgutierrez.stg_tasas_cambio_pais_anual"

#load table
stg_exchange = spark.read \
  .format("bigquery") \
  .option("table", table_exchange) \
  .load()
  
#select columns from table
raw_exchange = stg_exchange.select('Alpha_2_code','Alpha_3_code','Country_name','Year','currency','value')

#rename columns
raw_exchange = raw_exchange.withColumnRenamed('Alpha_2_code','country_code') \
                           .withColumnRenamed('Alpha_3_code','country_code_iso') \
                           .withColumnRenamed('Country_name','country_name') \
                           .withColumnRenamed('Year','year_rate') \
                           .withColumnRenamed('currency','currency_name') \
                           .withColumnRenamed('value','value_rate')
                           
#group by and select last value_rate 
df_group_rate = raw_exchange.select('country_code','year_rate','value_rate') \
        .groupBy('country_code',) \
        .agg(max('year_rate').alias('max_year'),last('value_rate').alias('value_exchange')) \
        .orderBy('country_code',asceding=False)

Mostramos la tasa de cambia actual para cada pais:

#Show row exchange
df_group_rate.show(5)
+------------+--------+--------------+
|country_code|max_year|value_exchange|
+------------+--------+--------------+
|          AR|    2020|      70.53917|
|          AT|    2020|      0.875506|
|          AU|    2020|      1.453085|
|          BE|    2020|      0.875506|
|          BG|    2020|      1.716333|
+------------+--------+--------------+

Ya tenemos una dataframe de productos y otra de tasa de cambios , hacemos un join.

#join dataframe df_clean_products_raw && df_exchange_group
df_merge_rows = df_group_rate.alias('rate') \
                .join(df_clean_products_raw.alias('price'), col('price.country') == col('rate.country_code'), "inner")
                          
#equivalente en dólares del precio de cada uno de los productos
df_raw_products=df_merge_rows.withColumn('app_sale_price_us', col('app_sale_price')/col('value_exchange'))


#rename columns
df_full_products = df_raw_products.withColumnRenamed('isprime','product_is_prime') \
                           .withColumnRenamed('app_sale_price_currency','product_price_currency') \
                           .withColumnRenamed('isbestseller','product_is_bestseller') \
                           .withColumnRenamed('clean_rate','product_rate') \
                           .withColumnRenamed('app_sale_price','product_price') \
                           .withColumnRenamed('country_code','product_country') \
                           .withColumnRenamed('app_sale_price_us','product_price_us')
                           
 #cast type                          
df_full_products = df_full_products.withColumn("product_price",df_full_products.product_price.cast(DoubleType()))  \
                                    .withColumn("product_rate",df_full_products.product_rate.cast(DoubleType())) \
                                    .withColumn("product_is_bestseller",df_full_products.product_is_bestseller.cast(StringType())) \
                                    .withColumn("product_is_prime",df_full_products.product_is_prime.cast(StringType())) 

Mostramos el resultado y esquema de dataframe:

#Show row exchange
df_full_products.show(2)
+----------+---------------------+----------------+----------------------+------------+-------------+---------------+------------------+
|product_id|product_is_bestseller|product_is_prime|product_price_currency|product_rate|product_price|product_country|  product_price_us|
+----------+---------------------+----------------+----------------------+------------+-------------+---------------+------------------+
|B07FXP7HVS|                 true|           false|                     €|         4.1|        18.19|             IT|20.776556642672926|
|B077T5RQF7|                 true|            true|                     €|         4.4|        50.48|             IT| 57.65808572414124|
|B074VMTP68|                 true|            true|                     €|         4.4|        29.99|             DE| 34.25447683967899|
|B00QHC01C2|                false|            true|                     €|         4.5|        29.72|             NL|33.946083750425466|
|B01GFJWHZ0|                 true|            true|                     €|         4.5|        21.43|             NL|24.477273713715267|
+----------+---------------------+----------------+----------------------+------------+-------------+---------------+------------------+
#Display Schema
df_full_products.printSchema()
root
 |-- product_id: string (nullable = true)
 |-- product_is_bestseller: string (nullable = true)
 |-- product_is_prime: string (nullable = true)
 |-- product_price_currency: string (nullable = true)
 |-- product_rate: double (nullable = true)
 |-- product_price: double (nullable = true)
 |-- product_country: string (nullable = true)
 |-- product_price_us: double (nullable = true)

Guardamos nuestro dataframe en Bigquery

df_full_products.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_products_standard_price") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('overwrite') \
  .save()

Back to Top

4.2 Creando tabla promedio de precio de productos

Creamos la tabla pr_products_avg_price , que consite en conocer en cuantos paises se ha vendido un producto y su precio promedio.

#calculated avg product_price_us && count product_country
df_avg_price=pr_products_price.select('product_id','product_price_us','product_country') \
        .groupBy('product_id') \
        .agg(mean('product_price_us').alias('product_avg_price_us'),countDistinct('product_country').alias('country_count')).sort('country_count', ascending=False)

Resultado:

#Show row products   
df_avg_price.show(3)
+----------+--------------------+-------------+
|product_id|product_avg_price_us|country_count|
+----------+--------------------+-------------+
|B00MNV8E0C|   24.73453126271878|            7|
|B007B9NV8Q|  20.634105088517593|            7|
|B00X4SCCFG|  17.748030361843725|            6|
+----------+--------------------+-------------+

Guardamos el dataframe en Bigquery

df_avg_price.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_products_avg_price") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('overwrite') \
  .save()

4.3 Creando tabla rango de precios de productos

Creamos la tabla pr_products_range_price , que consite en conocer en cuantos paises se ha vendido un producto y su precio promedio, asi como tambien su precio maximo , minimo y el pais.

#operations min_price && max_price
df_price_range=pr_products_price.select('product_id','product_country','product_price_us') \
        .groupBy('product_id') \
        .agg(min('product_price_us').alias('product_min_price'),first('product_country').alias('country_mix_price'), \
            max('product_price_us').alias('product_max_price'),last('product_country').alias('country_max_price'))  
            
#InnerJoin de dataframe df_avg_price && df_price_range
df_full_ranges = df_avg_price.alias('A').join(df_price_range.alias('B'), col('A.product_id') == col('B.product_id'), "inner") 

Verficamos que los datos sean correctos, buscando un producto en especifico:

#List product test
pr_products_price.filter(pr_products_price.product_id == "B00MNV8E0C").show(truncate=False)
+----------+---------------------+----------------+----------------------+------------+-------------+---------------+------------------+
|product_id|product_is_bestseller|product_is_prime|product_price_currency|product_rate|product_price|product_country|product_price_us  |
+----------+---------------------+----------------+----------------------+------------+-------------+---------------+------------------+
|B00MNV8E0C|true                 |false           |                      |4.3         |1146.0       |JP             |10.732891667660974|
|B00MNV8E0C|true                 |false           |€                     |4.6         |76.76        |DE             |87.67501307815138 |
|B00MNV8E0C|true                 |true            |₹                     |4.6         |1179.0       |IN             |15.909874809931647|
|B00MNV8E0C|true                 |false           |€                     |4.6         |13.79        |IT             |15.750891484467267|
|B00MNV8E0C|false                |true            |€                     |4.6         |11.69        |NL             |13.352278568050933|
|B00MNV8E0C|false                |false           |£                     |4.7         |11.49        |GB             |14.73076923076923 |
|B00MNV8E0C|true                 |true            |$                     |4.7         |14.99        |US             |14.99             |
+----------+---------------------+----------------+----------------------+------------+-------------+---------------+------------------+

#TEST PRODUCT
df_full_ranges.filter(df_full_ranges.product_id == "B00MNV8E0C").show(truncate=False)
+----------+-------------+--------------------+-----------------+-----------------+------------------+-----------------+
|product_id|country_count|product_avg_price_us|product_max_price|country_max_price|product_min_price |country_mix_price|
+----------+-------------+--------------------+-----------------+-----------------+------------------+-----------------+
|B00MNV8E0C|7            |24.73453126271878   |87.67501307815138|US               |10.732891667660974|JP               |
+----------+-------------+--------------------+-----------------+-----------------+------------------+-----------------+

Resultado:

#Show row mergedf_merge_rows
df_full_ranges.show(2)

#Show schema
df_full_ranges.printSchema()
+----------+-------------+--------------------+-----------------+-----------------+-----------------+-----------------+
|product_id|country_count|product_avg_price_us|product_max_price|country_max_price|product_min_price|country_mix_price|
+----------+-------------+--------------------+-----------------+-----------------+-----------------+-----------------+
|B09S5G7BXW|            1|                 0.0|              0.0|               US|              0.0|               US|
|B09S2RQ19K|            1|               99.99|            99.99|               US|            99.99|               US|
+----------+-------------+--------------------+-----------------+-----------------+-----------------+-----------------+
root
 |-- product_id: string (nullable = true)
 |-- country_count: long (nullable = false)
 |-- product_avg_price_us: double (nullable = true)
 |-- product_max_price: double (nullable = true)
 |-- country_max_price: string (nullable = true)
 |-- product_min_price: double (nullable = true)
 |-- country_mix_price: string (nullable = true)

Guardamos el dataframe en Bigquery

df_full_ranges.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_products_range_price") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('overwrite') \
  .save()

Back to Top

4.4 Creando tabla promedio de evaluación

Creando la tabla que consiste en obtener el promedio de evalucacion de cada producto..

#operations avg evaluate_rate products
df_product_rate=pr_products_price.select('product_id','product_country','product_rate') \
        .groupBy('product_id') \
        .agg(avg('product_rate').alias('product_avg_rate'),countDistinct('product_country').alias('country_count')) 

Resultado:

df_product_rate.show(2)
+----------+----------------+-------------+
|product_id|product_avg_rate|country_count|
+----------+----------------+-------------+
|B08J3QQ11H|             4.6|            1|
|9804370085|             4.8|            1|
+----------+----------------+-------------+

Guardando dataframe en Bigquery

df_product_rate.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_product_avg_rate") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('overwrite') \
  .save()

Back to Top

4.5 Creando tabla de clientes

Creando la tabla pr_clients desde stg_clients

#load table
raw_clients = spark.read \
  .format("bigquery") \
  .option("table", table_clients) \
  .load()
  
 #select columns 
 raw_clients = raw_clients.drop('_airbyte_ab_id','_airbyte_emitted_at','rowid')
 
 #renamed columns 
raw_clients = raw_clients.withColumnRenamed('id','client_id') \
                           .withColumnRenamed('nombre','client_name') \
                            .withColumnRenamed('direccion','client_address') \
                           .withColumnRenamed('email','client_email') \
                           .withColumnRenamed('isprime','client_is_prime') \
                           .withColumnRenamed('numero_tarjeta','client_credit_card') \
                           .withColumnRenamed('telefono','client_phone')

Resultado

raw_clients.show(10)
raw_clients.printSchema()

+--------------------+--------------------+-----------------+---------------+-----------------+-------------------+-------------+
|      client_address|        client_email|        client_id|client_is_prime|      client_name| client_credit_card| client_phone|
+--------------------+--------------------+-----------------+---------------+-----------------+-------------------+-------------+
| 811 Bond Route, ...|KPearson9610@outl...|310-309280-59-926|          false|Katherine Pearson|5395-1908-9123-4506|(310)052-0159|
| 7781 John Locks,...|DBell5699@hotmail...|323-462812-43-494|          false|      Denise Bell|5145-8058-2252-3801|(323)377-0937|
| 2649 Calhoun Ove...|DHunt4398@outlook...|510-658906-47-860|          false|       David Hunt|4962-3792-3311-9337|(510)911-5136|
| 8650 Villa Ridge...|Christine-Sta@out...|661-024332-95-247|          false|  Christine Stark|4551-7274-4304-6403|(661)700-7605|
| 8596 Morgan Spri...|GOconnor5032@hotm...|480-146888-22-806|          false|   George Oconnor|5450-4564-7903-8604|(480)395-3172|
| 873 Douglas Road...|[email protected]|702-421856-10-997|          false|      Tina Vargas|4113-8373-3475-5576|(702)027-5993|
| 2919 Harper Bypa...|BMartin3534@hotma...|623-179710-20-737|          false|    Brandi Martin|4142-5928-6392-1975|(623)635-2754|
| 9798 Charles Cro...|THebert9341@outlo...|415-292247-48-873|          false|     Tonya Hebert|5381-6017-6261-5547|(415)058-5014|
| 5358 Christopher...|Brianna-And@gmail...|520-858053-78-814|          false| Brianna Anderson|4141-4300-7963-7064|(520)263-2381|
| 7704 Sanchez Cir...|[email protected]|435-318881-49-746|          false|        Jo Taylor|5157-5473-7055-7677|(435)745-0343|
+--------------------+--------------------+-----------------+---------------+-----------------+-------------------+-------------+
root
 |-- client_address: string (nullable = true)
 |-- client_email: string (nullable = true)
 |-- client_id: string (nullable = true)
 |-- client_is_prime: string (nullable = true)
 |-- client_name: string (nullable = true)
 |-- client_credit_card: string (nullable = true)
 |-- client_phone: string (nullable = true)

Guardando datos en Bigquery:

raw_clients.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_clients") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('overwrite') \
  .save()

Back to Top

4.6 Creando tabla de compras

La tabla pr_compras es el resultado de la tablas stg_compras y stg_historico_compras

Cargamos los datos al dataframe

#name table compras
table_compras = "becade_mgutierrez.stg_compras" # or stg_historico_compras

#load table
raw_compras = spark.read \
  .format("bigquery") \
  .option("table", table_compras) \
  .load()
  
#selct columns  
df_raw_compras = raw_compras.select('fecha_compra','client_id','precio','product_id','cantidad','isprime')

#cast
df_raw_compras = df_raw_compras.withColumn("datetime", to_date("fecha_compra")) \
                               .withColumn("cantidad",df_raw_compras.cantidad.cast(IntegerType())) \
                                .withColumn("isprime",df_raw_compras.isprime.cast(StringType()))
#drop columns                               
df_raw_compras = df_raw_compras.drop('fecha_compra')

#rename columns
df_raw_compras = df_raw_compras.withColumnRenamed('datetime','purchase_date') \
                                .withColumnRenamed('cantidad','product_quantity') \
                                .withColumnRenamed('precio','product_price') \
                                .withColumnRenamed('isprime','client_is_prime')

Resultado:

df_raw_compras.show(2)
+-----------------+-------------+----------+----------------+-------------+
|        client_id|product_price|product_id|product_quantity|purchase_date|
+-----------------+-------------+----------+----------------+-------------+
|209-696678-32-117|       236.99|B00N69D6AS|               1|   2018-04-18|
|209-696678-32-117|       236.99|B00N69D6AS|               1|   2018-05-05|
+-----------------+-------------+----------+----------------+-------------+

Guardando datos en Bigquery:

df_raw_compras.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_compras") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('overwrite') \
  .save()

Back to Top

4.7 Creando tabla de compras anuales

Creando la tabla pr_compras_anuales que consite en obtener el año, venta total del año, promedio de ventas anuales, total de ordenes por año


#separando fecha en mes y año
df_new_sales= raw_compras.withColumn('month_sales',month(raw_compras.purchase_date)) \
                .withColumn('year_sales',year(raw_compras.purchase_date))
                
#obtneiendo la cantidad de clientes diferentes que comoraron al año
#group by year_sales | countDistinct client_id
df_ordenes_year = df_new_sales.select('year_sales','purchase_date','month_sales','client_id') \
        .groupBy('year_sales','purchase_date','client_id') \
        .agg(countDistinct('client_id').alias('total_compras')) \
        .sort(['year_sales', 'purchase_date'], ascending=True)

#obteniendo el total de ordenes por año
#sum ordenes by client_id | groupby year_sales | count total_compras
sum_ordenes_year = df_ordenes_year.select('year_sales','total_compras') \
        .groupBy('year_sales') \
        .agg(count('total_compras').alias('total_compras')) \
        .sort('year_sales', ascending=True)
   
#obteniendo el promedio de venta por año y el total de venta por año
#groupBy year_sales | sum product_price | avg product_price
df_sales = df_new_sales.select('year_sales','month_sales','product_price','client_id') \
        .groupBy('year_sales') \
        .agg(sum('product_price').alias('venta_total_year'), \
             avg('product_price').alias('avg_venta_mensual')) \
         .sort('year_sales', ascending=True)
         
#InnerJoin df_sales && sum_ordenes_year
full_table_year = df_sales.alias('A').join(sum_ordenes_year.alias('B'), col('A.year_sales') == col('B.year_sales'), "inner") 

Resultado

full_table_year.show()
+----------+--------------------+------------------+-------------+
|year_sales|    venta_total_year| avg_venta_mensual|total_compras|
+----------+--------------------+------------------+-------------+
|      2010|1.8495069750010703E7|104.32923661415366|        32717|
|      2011|1.8595521430009063E7|104.14100184255835|        32746|
|      2012| 1.891395686000886E7|104.80852955197692|        32899|
|      2013| 1.855743942000939E7|103.84979669274121|        32733|
|      2014|1.8657688250007752E7| 105.1688391664802|        32713|
|      2015| 1.872384307000746E7|104.68142492945773|        32957|
|      2016| 1.867335246000713E7|104.62258287908166|        32854|
|      2017| 1.872034121000714E7|104.20509554746833|        32753|
|      2018|1.8488971880010866E7|104.16204820233499|        32724|
|      2019|1.8643064240007747E7|104.42012243827817|        32755|
|      2020|1.8892210430008642E7|104.74258421676042|        32913|
+----------+--------------------+------------------+-------------+

Guardando dataframe en Bigquery

full_table_year.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_compras_anuales") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('overwrite') \
  .save()

Back to Top

4.8 Creando tabla de compras mensuales

Creamos la tabla pr_compras_mensuales que consite en obtener el año y mes, venta total y total de ordenes del mes y la venta total mismo mes pero del año anterior.

###compras por Mes
df_ordenes_month = df_new_sales.select('year_sales','purchase_date','month_sales','client_id') \
        .groupBy('year_sales','month_sales','purchase_date','client_id') \
        .agg(countDistinct('client_id').alias('total_compras')) \
        .sort(['year_sales', 'purchase_date'], ascending=True)
 
 sum_ordenes_month = df_ordenes_month.select('year_sales','month_sales','total_compras') \
        .groupBy('year_sales','month_sales') \
        .agg(count('total_compras').alias('total_compras_mes')) \
        .sort(['year_sales','month_sales'], ascending=True)
        
df_month = df_new_sales.select('year_sales','month_sales','product_price') \
        .groupBy('year_sales','month_sales') \
        .agg(sum('product_price').alias('venta_total_mes')) \
        .sort(['year_sales','month_sales'], ascending=True)
 
 df_month_raw = df_month.withColumn('venta_total_mes_anterior',lag(df_month['venta_total_mes']).over(Window.orderBy("month_sales","year_sales")))
 
 df_month_raw= df_month_raw.na.fill(value=0,subset=["venta_total_mes_anterior"])
 
 full_table_month = df_month_raw.alias('A').join(sum_ordenes_month.alias('B'), \
                (col('A.month_sales') == col('B.month_sales')) & (col('A.year_sales') == col('B.year_sales')) , "inner") 
full_table_month.show(13)
+----------+-----------+------------------+------------------------+-----------------+
|year_sales|month_sales|   venta_total_mes|venta_total_mes_anterior|total_compras_mes|
+----------+-----------+------------------+------------------------+-----------------+
|      2010|          1|1241494.2399999024|                     0.0|             2618|
|      2011|          1|1220903.2599999004|      1241494.2399999024|             2586|
|      2012|          1|1207546.9799999103|      1220903.2599999004|             2577|
|      2013|          1|1229377.7699999062|      1207546.9799999103|             2569|
|      2014|          1|1288639.0199998915|      1229377.7699999062|             2602|
|      2015|          1|1270299.0199999036|      1288639.0199998915|             2630|
|      2016|          1|1345551.9299998868|      1270299.0199999036|             2619|
|      2017|          1|1145680.2099999334|      1345551.9299998868|             2529|
|      2018|          1|1198162.2299999034|      1145680.2099999334|             2555|
|      2019|          1|1221042.8499999032|      1198162.2299999034|             2605|
|      2020|          1|1253193.3999999021|      1221042.8499999032|             2610|
|      2010|          2|1230857.7399999062|      1253193.3999999021|             2406|
|      2011|          2|1168152.5199999078|      1230857.7399999062|             2401|
+----------+-----------+------------------+------------------------+-----------------+
full_table_month.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_compras_mensuales") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('overwrite') \
  .save()

Back to Top

5 Tabla de hechos

Desempeño de ventas por producto a nivel anual, es decir la cantidad total obtenida de la venta de ese producto por año..

● La cantidad de usuarios que compraron este producto en ese año.

● La cantidad de unidades que se vendieron de ese producto durante ese año.

● El evaluation_rating actual del producto

#separamos la fecha en años
df_fact_raw= raw_compras.withColumn('purchase_year',year(raw_compras.purchase_date)) 

#groupby by purchase_year and product_id | sum product_price | countDistinct client_id | sum product_quantity
df_fact_sales = df_fact_raw.select('*') \
                .groupBy('purchase_year','product_id') \
                .agg(sum('product_price').alias('purchase_sales'),
                     countDistinct('client_id').alias('client_quantity'), 
                     sum('product_quantity').alias('product_quantity_sales') ) \
                .sort(['purchase_year', 'product_id'], ascending=False)
                
#InnerJoin df_fact_sales &&  raw_rate
full_table_fact = df_fact_sales.alias('A').join(raw_rate.alias('B'), \
                (col('A.product_id') == col('B.product_id')) , "inner") 
                 

Resultado:

full_table_fact.show(13)
+-------------+----------+------------------+---------------+----------------------+----------------+
|purchase_year|product_id|    purchase_sales|client_quantity|product_quantity_sales|product_avg_rate|
+-------------+----------+------------------+---------------+----------------------+----------------+
|         2020|B09FM7CKG1| 56771.82000000001|            100|                  1959|             3.6|
|         2020|B09FCXXGT5|112020.12000000007|            100|                  3749|             4.8|
|         2020|B09BNK4592|111496.56000000007|            100|                  1862|             4.7|
|         2020|B098RKWHHZ| 695430.1299999999|            100|                  1987|             4.8|
|         2020|B098P1M628| 551117.1599999984|            100|                 11484|             3.9|
|         2020|B097Y38X79| 537275.8800000008|            100|                 11532|             3.9|
|         2020|B0975P2RBR| 79900.01999999999|            100|                  1998|             3.6|
|         2020|B0931NN4PR|          195552.0|            100|                  2016|             4.8|
|         2020|B0914YGQSH|117304.92000000004|            100|                  1959|             4.6|
|         2020|B08ZS9PQ78| 87760.38000000005|            100|                  9762|             4.6|
|         2020|B08Z11QHBG| 686016.0399999997|            100|                  1966|             4.8|
|         2020|B08XXWHLQF|49484.959999999985|            100|                  1904|             4.1|
|         2020|B08X2K6B1Z| 78747.40000000004|            100|                  1802|             4.8|
+-------------+----------+------------------+---------------+----------------------+----------------+

Guardando dataframe en Bigquery

raw_compras_historico.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_fact_compras") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('append') \
  .save()

Back to Top

6. Información de cargas incrementales desde BigQuery

Desarrolla un proceso que lleve los datos incrementales de manera diaria a la tabla de compras en la capa de staging de BigQuery para tener un respaldo de los datos tal cual llegan de la fuente, la temporalidad de estos procesos se definen a continuación:

1- Extraer diariamente los datos del día anterior y llevarlos a la capa de staging (stg_compras) de BigQuery.

2- Una vez cada primero del mes podrías realizar la limpieza y transformaciones necesarias para obtener los datos que debes insertar en la tabla de compras_mensuales aplicando esto a todas las compras con fecha del mes anterior.

3- Aplicar una lógica similar a la anterior pero de manera anual en vez de mensual.

La única información que se verá afectada por estos datos incrementales serán los que tengan que ver con la tabla de compras

Back to Top

6.1 Carga de datos de compras diarias

# # #filter data previus day
raw_previus_day= raw_compras.filter(raw_compras.fecha_compra == date_sub(current_date(),1))

raw_previus_day = raw_previus_day.select('fecha_compra','client_id','precio','product_id','cantidad','isprime')

raw_previus_day = raw_previus_day.withColumn("cantidad",raw_previus_day.cantidad.cast(IntegerType())) \
                                .withColumn("isprime",raw_previus_day.isprime.cast(StringType()))
                                
#rename columns
raw_previus_final = raw_previus_day.withColumnRenamed('fecha_compra','purchase_date') \
                                .withColumnRenamed('cantidad','product_quantity') \
                                .withColumnRenamed('precio','product_price') \
                               .withColumnRenamed('isprime','client_is_prime')  


Guardamos dataset

raw_previus_day.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_compras") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('append') \
  .save()

Back to Top

6.2 Calculo de compras anuales

df_sales_current_year = raw_current_year.filter(  (date_trunc("month", col("purchase_date")) != date_trunc("month", current_date())) &
                                                (date_trunc("year", col("purchase_date")) == date_trunc("year", current_date())))
 
 
df_new_sales= df_sales_current_year.withColumn('month_sales',month(df_sales_current_year.purchase_date)) \
                .withColumn('year_sales',year(df_sales_current_year.purchase_date))

df_ordenes_year = df_new_sales.select('year_sales','purchase_date','month_sales','client_id') \
        .groupBy('year_sales','purchase_date','client_id') \
        .agg(countDistinct('client_id').alias('total_compras')) \
        .sort(['year_sales', 'purchase_date'], ascending=True)
        
 sum_ordenes_year = df_ordenes_year.select('year_sales','total_compras') \
        .groupBy('year_sales') \
        .agg(count('total_compras').alias('total_compras')) \
        .sort('year_sales', ascending=True)       

df_sales = df_new_sales.select('year_sales','month_sales','product_price','client_id') \
        .groupBy('year_sales') \
        .agg(sum('product_price').alias('venta_total_year'), \
             avg('product_price').alias('avg_venta_mensual')) \
         .sort('year_sales', ascending=True)
         
#InnerJoin
full_table_year = df_sales.alias('A').join(sum_ordenes_year.alias('B'), col('A.year_sales') == col('B.year_sales'), "inner") 

Guardamos dataframe *****

full_table_year.write \
  .format("bigquery") \
  .option("table","becade_mgutierrez.pr_compras_anuales") \
  .option("temporaryGcsBucket", "amazon_magdielgutierrez") \
  .mode('append') \
  .save()

Back to Top

6.3 Calculo de compras mensuales

?

Back to Top

Releases

No releases published

Packages

No packages published