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 ? 

Here’s a well-formatted version of your text:

---

Your approach is on the right track! Determining the product that has been sold in the most orders can be achieved efficiently using the `df_sales` DataFrame. However, whether you need to join with the `df_products` DataFrame depends on the level of detail you want in your final result.

### 1. Without Joining (Using Only `df_sales`):
If your goal is to identify the `product_id` with the highest number of orders, you can accomplish this directly using the `df_sales` DataFrame without any joins. This approach is simpler and more efficient, especially with large datasets, as it avoids the overhead of joining tables.

**Example Code:**

```python
from pyspark.sql import functions as F

# Count the number of orders for each product_id
product_order_counts = df_sales.groupBy("product_id") \
    .agg(F.count("order_id").alias("order_count"))

# Find the product_id with the maximum number of orders
most_sold_product = product_order_counts.orderBy(F.desc("order_count")).limit(1)

# Display the result
print("Product ID sold in the most orders:")
most_sold_product.show()
```

**Output Example:**

```
+----------+-----------+
|product_id|order_count|
+----------+-----------+
|      P010|       1500|
+----------+-----------+
```

### 2. With Joining (Including `product_name`):
If you want more context, such as the `product_name`, joining with the `df_products` DataFrame is beneficial. This provides a more informative result, making it easier to interpret which specific product is the top seller.

**Example Code:**

```python
from pyspark.sql import functions as F

# Count the number of orders for each product_id
product_order_counts = df_sales.groupBy("product_id") \
    .agg(F.count("order_id").alias("order_count"))

# Join with df_products to get product_name
most_sold_product = product_order_counts.join(df_products, on="product_id") \
    .orderBy(F.desc("order_count")) \
    .limit(1)

# Display the result
print("Product sold in the most orders:")
most_sold_product.select("product_id", "product_name", "order_count").show()
```

**Output Example:**

```
+----------+------------+-----------+
|product_id|product_name|order_count|
+----------+------------+-----------+
|      P010| SuperWidget|       1500|
+----------+------------+-----------+
```

### 3. Optimizing the Query:
To ensure optimal performance, especially with large datasets, consider the following best practices:

- **Use Selective Columns**: Only select the columns you need for your analysis. This reduces the amount of data shuffled and processed.

    ```python
    product_order_counts = df_sales.select("product_id", "order_id") \
        .groupBy("product_id") \
        .agg(F.count("order_id").alias("order_count"))
    ```

- **Cache Intermediate DataFrames**: If you plan to reuse a DataFrame multiple times, caching it can save computation time.

    ```python
    product_order_counts.cache()
    ```

- **Broadcast Joins**: If `df_products` is small enough to fit into memory, broadcasting it can speed up the join operation.

    ```python
    from pyspark.sql.functions import broadcast

    most_sold_product = product_order_counts.join(
        broadcast(df_products), on="product_id"
    ).orderBy(F.desc("order_count")).limit(1)
    ```

- **Partitioning**: Ensure your DataFrames are appropriately partitioned to optimize parallel processing.

    ```python
    spark.conf.set("spark.sql.shuffle.partitions", "200")  # Adjust based on your cluster
    ```

### 4. Complete Example Combining All Steps:
Here’s a comprehensive example incorporating schema definitions, DataFrame creations, and the process to find the most sold product with optimization considerations:

```python
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.functions import broadcast

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

# Define schemas
products_schema = StructType([
    StructField("product_id", StringType(), True),
    StructField("product_name", StringType(), True),
    StructField("price", DoubleType(), True)
])

sales_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("seller_id", StringType(), True),
    StructField("date", StringType(), True),
    StructField("num_pieces_sold", IntegerType(), True),
    StructField("bill_raw_text", StringType(), True)
])

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
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)

# Cache df_sales if it's reused frequently
df_sales.cache()

# Count orders per product_id
product_order_counts = df_sales.groupBy("product_id") \
    .agg(F.count("order_id").alias("order_count"))

# Broadcast join if df_products is small
most_sold_product = product_order_counts.join(
    broadcast(df_products), on="product_id"
).orderBy(F.desc("order_count")).limit(1)

# Show the most sold product with details
most_sold_product.select("product_id", "product_name", "order_count").show()
```

### 5. Final Thoughts:
- **Simpler vs. Informative**: Using only `df_sales` is sufficient for counting orders, but joining with `df_products` adds valuable context.
- **Efficiency**: Avoid unnecessary joins when possible to maintain query efficiency, especially with large datasets.
- **Flexibility**: Structuring your queries to allow easy inclusion or exclusion of joins provides flexibility based on your analysis needs.

Feel free to implement the approach that best suits your specific requirements. If you need further assistance or have additional questions, don’t hesitate to ask!


Spark SQL :  Manipulation des vues et des tables : 
Créer trois tables « Products », « Sales » et « Sellers » à partir de ces data frames.
—> now we move to the next “products” “sales” and “seller” as tables using delta format as open table format oky 

Give the code to do it oky