Q- give i use colab and installed this 


Collecting pyspark
  Downloading pyspark-3.5.3.tar.gz (317.3 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 317.3/317.3 MB 4.1 MB/s eta 0:00:00
  Preparing metadata (setup.py) ... done
Requirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from pyspark) (0.10.9.7)
Building wheels for collected packages: pyspark
  Building wheel for pyspark (setup.py) ... done
  Created wheel for pyspark: filename=pyspark-3.5.3-py2.py3-none-any.whl size=317840625 sha256=449954b332f613db9b0d5464c80c37b7ec1e76bd2988bf2854f52efbeb49f64b
  Stored in directory: /root/.cache/pip/wheels/1b/3a/92/28b93e2fbfdbb07509ca4d6f50c5e407f48dce4ddbda69a4ab
Successfully built pyspark
Installing collected packages: pyspark
Successfully installed pyspark-3.5.3

Q- my question it’s possible to use spark sql here ? 
Yes 

Objectif de TP :
Comprendre et appliquer des concepts avancés de Spark SQL.
Optimiser les requêtes à l'aide des techniques de gestion de la mémoire et des jointures.
Utiliser des fonctions UDF et UDAF pour des traitements spécifiques.
Gérer des jeux de données massifs avec des stratégies de partitionnement efficaces.

Etude de cas : Analyse des ventes
Soit le jeu de données qui se compose de trois tables issues de la base de données d'une boutique, avec les produits, les ventes et les vendeurs. La figure suivante montre comment les tables peuvent être connectées :


Expert A: Let's analyze this entity-relationship diagram. It appears to represent a sales system with three main entities: Products, Sales, and Sellers. What's your initial impression?

Expert B: I agree. The diagram shows a clear relationship between these three entities. The Sales table acts as a bridge between Products and Sellers, which is typical in a transactional database design. Let's break down each entity.

Expert A: Good point. Starting with the Products table, we see three attributes: product_id, product_name, and price. The product_id is likely the primary key here.

Expert B: Correct. Moving to the Sellers table, we have seller_id, seller_name, and an interesting attribute called daily_target. This suggests that each seller might have a daily sales goal to meet.

Expert A: Interesting observation. The Sales table is the most complex, containing six attributes: order_id, product_id, seller_id, date, num_pieces_sold, and bill_raw_text. The order_id is probably the primary key for this table.

Expert B: Yes, and notice how product_id and seller_id in the Sales table are foreign keys, linking to the Products and Sellers tables respectively. This allows us to track which product was sold by which seller in each transaction.

Expert A: Exactly. The date field enables time-based analysis, while num_pieces_sold likely represents the quantity of products sold in each transaction. The bill_raw_text field is intriguing - it might contain the full text of the sales receipt or invoice.

Expert B: That's a good guess. Including raw text data can be useful for auditing or for extracting additional information that doesn't fit into the structured fields.

Expert A: Overall, this schema seems designed to answer questions like: What products are being sold? Who is selling them? How many units are being sold daily? Are sellers meeting their daily targets?

Expert B: Agreed. It's a straightforward yet effective design for tracking sales performance. However, it might benefit from additional tables or fields depending on specific business needs, such as customer information or more detailed product categories.

Expert A: Good point. This diagram provides a solid foundation for a basic sales tracking system, with room for expansion as the business requirements evolve.

First part 

“Exercise Objectives:
Loading and Creating DataFrames:
Create three DataFrames from the CSV files "Products", "Sales", and "Sellers".
Define their schemas to prevent automatic type inference.
Display the first five rows and the schema of each DataFrame.
“


Q- Create three DataFrames from the CSV files "Products", "Sales", and "Sellers".
path_products="/content/products.csv"
path_sales="/content/sales.csv"
path_sellers="/content/sellers.csv"
"""
now create a data_frame called df_products , df_sales , df_sellers
"""

Do it for me 
Here’s how you can create the three DataFrames (`df_products`, `df_sales`, and `df_sellers`) from the CSV files in PySpark, while explicitly defining the schema to prevent automatic type inference:

### Step 1: Define the Schemas for Each DataFrame
To prevent automatic type inference, we define the schema using `StructType` and `StructField`.

```python
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType
from pyspark.sql import SparkSession

# Start a Spark session
spark = SparkSession.builder.appName("SalesAnalysis").getOrCreate()

# Define schema for Products DataFrame
products_schema = StructType([
    StructField("product_id", StringType(), True),
    StructField("product_name", StringType(), True),
    StructField("price", DoubleType(), True)
])

# Define schema for Sales DataFrame
sales_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("seller_id", StringType(), True),
    StructField("date", StringType(), True),  # You can change it to DateType if you have date format control
    StructField("num_pieces_sold", IntegerType(), True),
    StructField("bill_raw_text", StringType(), True)
])

# Define schema for Sellers DataFrame
sellers_schema = StructType([
    StructField("seller_id", StringType(), True),
    StructField("seller_name", StringType(), True),
    StructField("daily_target", IntegerType(), True)
])

# File paths
path_products = "/content/products.csv"
path_sales = "/content/sales.csv"
path_sellers = "/content/sellers.csv"

# Create DataFrames by reading the CSV files with the defined schema
df_products = spark.read.csv(path_products, header=True, schema=products_schema)
df_sales = spark.read.csv(path_sales, header=True, schema=sales_schema)
df_sellers = spark.read.csv(path_sellers, header=True, schema=sellers_schema)

# Display first five rows of each DataFrame and their schema
print("Products DataFrame:")
df_products.show(5)
df_products.printSchema()

print("Sales DataFrame:")
df_sales.show(5)
df_sales.printSchema()

print("Sellers DataFrame:")
df_sellers.show(5)
df_sellers.printSchema()
```

### Explanation:
1. **Schemas**: We manually define the schema for each DataFrame using `StructType` and `StructField` to specify the types for each column.
2. **CSV Loading**: Each CSV file is read into a DataFrame using `spark.read.csv()`, with `header=True` to indicate that the CSV has headers.
3. **Display**: We show the first five rows using `show(5)` and print the schema using `printSchema()`.

### Output Example:

- **Products DataFrame**:
  ```
  +----------+------------+-----+
  |product_id|product_name|price|
  +----------+------------+-----+
  |      P001|    Product1| 10.0|
  |      P002|    Product2| 20.0|
  |      P003|    Product3| 15.5|
  |      P004|    Product4|  7.5|
  |      P005|    Product5| 30.0|
  +----------+------------+-----+
  ```

- **Schema**:
  ```
  root
   |-- product_id: string (nullable = true)
   |-- product_name: string (nullable = true)
   |-- price: double (nullable = true)
  ```

This code snippet will create and load the DataFrames and provide an overview of their structure and data.

Q- DataFrame Manipulation:
Find out how many orders, how many products, and how many sellers are in the data.
 So the question ask us how mafny orders 
→ we can solve it using the equivante count of sql world in the sparkSql 
Same think for the next ones 

Determine the products that have been sold at least once.
Q- so the next question is about computing the products that have been sold at least once 

The desired table will be products 
We fetch product_id , product_name 
Then also use the aggregation function to compute the number of saled product and also make alias of it for readability 

Then filter based only  products that have been sold at least once oky 

→ translate my thinking process into actual code and opitmze always 

from pyspark.sql import functions as F

# Join products and sales DataFrames, then aggregate and filter
products_sold = df_products.join(
    df_sales, 
    df_products.product_id == df_sales.product_id, 
    "left"
).groupBy(
    df_products.product_id, 
    df_products.product_name
).agg(
    F.count(df_sales.order_id).alias("sales_count")
).filter(
    F.col("sales_count") > 0
).select(
    "product_id", 
    "product_name", 
    "sales_count"
).orderBy(
    F.desc("sales_count")
)

# Show the results
print("Products sold at least once:")
products_sold.show()

# Get the count of products sold at least once
products_sold_count = products_sold.count()
print(f"Number of products sold at least once: {products_sold_count}")


Determine the product that has been sold in the most orders.
So we need to find the product that have most solded (aka the most desired product) 

→ Q- do we need to join the two tables yes or noo ? 
I think no because we can use “sales” table only here oky 

Give your feedback here ?