The company ‘X’ has recently started tracking sales data for its products and has stored the information in a MySQL database. The database contains information such as quantity ordered for each product, price of each product on the date of Order, name of the product, Status(meaning the Product has been shipped or not), Date on which the product was ordered. The company ‘X’ would like to have a quick and easy way to analyze the sales data to gain insights into the performance of each product. Objective: Develop a distributed software application using Spark that will extract sales data from the MySQL database, perform data analysis tasks on the data, and display the results in a HTML file. The application should perform the following tasks: Create MySQL Database: Take a sample_data.csv file as input and store it in a MySQL database. This MySQL database should be stored a docker container. Write the Dockerfile which creates this database and stores the data present in the sample_data.csv file as a table. Design the Data Processing Application: Write a python script which imports the data from MySQL database and performs data processing task such as: Calculates total sale for each time every month. Ranks item based on total units sold every month. The data processing task is to be performed using PySpark. The input data should be stored in the form of Dataframes and the output should be stored in the csv files. For both the task (1) and (2), two separate csv files should be produced as output. Also, write a python script which takes these two csv files as input and stores them in a single HTML file. The HTML file needs to be beautified using CSS and javascript. This complete application(data processing task and storing in HTML file) is to be stored in a separate Docker container. Write a Dockerfile which stores in this application and also installs PySpark and all it’s dependencies. The Dockerfile should also set up all the environment variable so that PySpark and all the essential libraries are fully functional. Finally, write a docker-compose.yml file which establishes a connection between the Docker container which stores the data in MySQL and the Docker container which stores the python application.
Solution:
To achieve the objective mentioned above, we can follow the following steps:
Step 1: Create MySQL Database
First, we need to create a docker container that will store our MySQL database. We can do this by writing a Dockerfile that sets up a MySQL server and copies our sample_data.csv file into it. The Dockerfile for this would look something like this:
Dockerfile
FROM mysql:latest
ENV MYSQL_DATABASE sales_data
ENV MYSQL_ROOT_PASSWORD password
COPY sample_data.csv /docker-entrypoint-initdb.d/
EXPOSE 3306In this Dockerfile, we have specified the base image as mysql:latest and set up environment variables for the MySQL database. We have also copied our sample_data.csv file into the /docker-entrypoint-initdb.d/ directory, which is a directory where MySQL looks for SQL scripts to run on initialization.
We can build this Dockerfile and run the container using the following commands:
kotlin
docker build -t mysql-sales-data .
docker run -d -p 3306:3306 --name sales-data-container mysql-sales-dataThis will create and run the container that will store our MySQL database.
Step 2: Design the Data Processing Application
Next, we need to write a Python script that will extract the sales data from the MySQL database and perform data analysis tasks on it using PySpark. We can use the PyMySQL library to connect to our MySQL database and create a PySpark DataFrame from the data.
Here's an example Python script that calculates total sale for each time every month and ranks items based on total units sold every month:
python
import pyspark.sql.functions as F
from pyspark.sql import SparkSession
# create SparkSession
spark = SparkSession.builder.appName("sales_analysis").getOrCreate()
# create DataFrame from MySQL database
jdbc_url = "jdbc:mysql://localhost:3306/sales_data"
connection_properties = {
"user": "root",
"password": "password",
"driver": "com.mysql.cj.jdbc.Driver"
}
sales_df = spark.read.jdbc(jdbc_url, table="sales", properties=connection_properties)
# calculate total sale for each month
monthly_sales_df = sales_df \
.groupBy(F.month("order_date").alias("month"), F.year("order_date").alias("year")) \
.agg(F.sum("quantity_ordered" * "price_each").alias("total_sales")) \
.orderBy(F.desc("total_sales"))
# rank items based on total units sold each month
monthly_item_sales_df = sales_df \
.groupBy(F.month("order_date").alias("month"), F.year("order_date").alias("year"), "product_name") \
.agg(F.sum("quantity_ordered").alias("total_units_sold")) \
.orderBy(F.desc("total_units_sold"))
# write output to csv files
monthly_sales_df.write.mode("overwrite").csv("monthly_sales.csv")
monthly_item_sales_df.write.mode("overwrite").csv("monthly_item_sales.csv")In this script, we first create a SparkSession and use it to create a PySpark DataFrame from the MySQL database using the pyspark.sql.DataFrameReader.jdbc method. We then use PySpark functions to perform the data analysis tasks, such as grouping by month and year and aggregating the total sales and total units sold.
Finally, we write the output to two separate CSV files using the pyspark.sql.DataFrameWriter.csv method.
Step 3: Store the Application in a Docker Container
Now that we have our Python script that performs the data analysis tasks, we can write a Dockerfile that sets up a container to run this script along with PySpark and all its dependencies. Here
Perform the following task as well: write a python script which takes these two csv files as input and stores them in a single HTML file. The HTML file needs to be beautified using CSS and javascript.
Sure, here is a Python script that takes two CSV files as input, reads them, and generates a single HTML file with a beautiful tabular representation of the data using CSS and JavaScript:
python
import pandas as pd
# Read the CSV files
total_sales_df = pd.read_csv('total_sales.csv')
ranked_items_df = pd.read_csv('ranked_items.csv')
# Create HTML table for total sales
total_sales_table = total_sales_df.to_html(classes="table table-striped", index=False)
# Create HTML table for ranked items
ranked_items_table = ranked_items_df.to_html(classes="table table-striped", index=False)
# Combine the tables into a single HTML page
html_content = f"""
<html>
<head>
<title>Sales Analysis Report</title>
<link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/4.0.0/css/bootstrap.min.css"
integrity="sha384-Gn5384xqQ1aoWXA+058RXPxPg6fy4IWvTNh0E263XmFcJlSAwiGgFAW/dAiS6JXm"
crossorigin="anonymous">
</head>
<body>
<h1>Sales Analysis Report</h1>
<h2>Total Sales</h2>
{total_sales_table}
<h2>Ranked Items</h2>
{ranked_items_table}
<script src="https://code.jquery.com/jquery-3.2.1.slim.min.js"
integrity="sha384-KJ3o2DKtIkvYIK3UENzmM7KCkRr/rE9/Qpg6aAZGJwFDMVNA/GpGFF93hXpG5KkN"
crossorigin="anonymous"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/popper.js/1.12.9/umd/popper.min.js"
integrity="sha384-ApNbgh9B+Y1QKtv3Rn7W3mgPxhU9K/ScQsAP7hUibX39j7fakFPskvXusvfa0b4Q"
crossorigin="anonymous"></script>
<script src="https://maxcdn.bootstrapcdn.com/bootstrap/4.0.0/js/bootstrap.min.js"
integrity="sha384-JZR6Spejh4U02d8jOt6vLEHfe/JQGiRRSQQxSfFWpi1MquVdAyjUar5+76PVCmYl"
crossorigin="anonymous"></script>
</body>
</html>
"""
# Write the HTML file
with open('sales_analysis_report.html', 'w') as f:
f.write(html_content)In this script, we are using the pandas library to read the CSV files and convert them into HTML tables. We are then combining the two tables into a single HTML page and using Bootstrap CSS and JavaScript to style the tables and make them responsive. Finally, we are writing the generated HTML content to a file called sales_analysis_report.html.
Note: This script assumes that the two CSV files are named total_sales.csv and ranked_items.csv, and are in the same directory as the script.
For the below question write all the Dockerfiles which are required in the Problem: The company ‘X’ has recently started tracking sales data for its products and has stored the information in a MySQL database. The database contains information such as quantity ordered for each product, price of each product on the date of Order, name of the product, Status(meaning the Product has been shipped or not), Date on which the product was ordered. The company ‘X’ would like to have a quick and easy way to analyze the sales data to gain insights into the performance of each product. Objective: Develop a distributed software application using Spark that will extract sales data from the MySQL database, perform data analysis tasks on the data, and display the results in a HTML file. The application should perform the following tasks: Create MySQL Database: Take a sample_data.csv file as input and store it in a MySQL database. This MySQL database should be stored a docker container. Write the Dockerfile which creates this database and stores the data present in the sample_data.csv file as a table. Design the Data Processing Application: Write a python script which imports the data from MySQL database and performs data processing task such as: Calculates total sale for each time every month. Ranks item based on total units sold every month. The data processing task is to be performed using PySpark. The input data should be stored in the form of Dataframes and the output should be stored in the csv files. For both the task (1) and (2), two separate csv files should be produced as output. Also, write a python script which takes these two csv files as input and stores them in a single HTML file. The HTML file needs to be beautified using CSS and javascript. This complete application(data processing task and storing in HTML file) is to be stored in a separate Docker container. Write a Dockerfile which stores in this application and also installs PySpark and all it’s dependencies. The Dockerfile should also set up all the environment variable so that PySpark and all the essential libraries are fully functional. Finally, write a docker-compose.yml file which establishes a connection between the Docker container which stores the data in MySQL and the Docker container which stores the python application.
Dockerfile for MySQL Database:
sql
FROM mysql:latest
ENV MYSQL_ROOT_PASSWORD=root
ENV MYSQL_DATABASE=sales_data
COPY sample_data.csv /docker-entrypoint-initdb.d/
EXPOSE 3306Dockerfile for Data Processing Application:
bash
FROM python:3.8
ENV PYSPARK_PYTHON=python3
ENV SPARK_HOME=/spark
ENV PYTHONPATH=${SPARK_HOME}/python:${PYTHONPATH}
RUN apt-get update && \
apt-get install -y openjdk-11-jdk-headless && \
apt-get clean && \
rm -rf /var/lib/apt/lists/*
RUN pip install pyspark
COPY data_processing_script.py /
COPY beautify_html.py /
COPY style.css /
COPY script.js /
ENTRYPOINT ["python", "data_processing_script.py"]docker-compose.yml:
yaml
version: '3'
services:
mysql:
build: ./mysql/.
ports:
- 3306:3306
data_processing:
build: ./data_processing/.
volumes:
- ./output:/output
- ./html:/html
depends_on:
- mysqlWrite the .sql code for creating a table inside the Database and importing data from sample_data.csv file to that table.
Assuming that the MySQL database has been created, and the sample_data.csv file is located in the same directory as the Dockerfile for the MySQL container, the following SQL code can be used to create a table and import data from the CSV file:
sql
USE sales_data;
CREATE TABLE sales (
order_date DATE,
product_name VARCHAR(255),
quantity_ordered INT,
price_each FLOAT,
order_status VARCHAR(255)
);
LOAD DATA INFILE '/docker-entrypoint-initdb.d/sample_data.csv'
INTO TABLE sales
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
IGNORE 1 ROWS;This code creates a table called "sales" with five columns: "order_date", "product_name", "quantity_ordered", "price_each", and "order_status". It then uses the LOAD DATA INFILE command to import the data from the CSV file into the table. The "IGNORE 1 ROWS" command is used to skip the first row of the CSV file, which contains the column headers.
3 / 3
Modify the below Dockerfile code for mysql service such that the above .sql file is incorporated inside the container. FROM mysql:latest ENV MYSQL_DATABASE sales_data ENV MYSQL_ROOT_PASSWORD password COPY sample_data.csv /docker-entrypoint-initdb.d/ EXPOSE 3306
To incorporate the above .sql file inside the container, we need to modify the Dockerfile for the mysql service as follows:
FROM mysql:latest
ENV MYSQL_DATABASE sales_data ENV MYSQL_ROOT_PASSWORD password
COPY sample_data.csv /docker-entrypoint-initdb.d/ COPY create_table.sql /docker-entrypoint-initdb.d/
EXPOSE 3306
The above Dockerfile adds a new COPY command to copy the create_table.sql file to the /docker-entrypoint-initdb.d/ directory inside the container along with the sample_data.csv file. This will ensure that the table is created with the required schema and the data from the sample_data.csv file is imported into the table.
Modify the below snippet of code from data_processing_script.py file such that it imports data from MySQL database without using JDBC Driver. # create DataFrame from MySQL database jdbc_url = "jdbc:mysql://localhost:3306/sales_data" connection_properties = { "user": "root", "password": "password", "driver": "com.mysql.cj.jdbc.Driver" }
To import data from MySQL database without using JDBC Driver, we can use PySpark's built-in MySQL connector. Here's the modified code:
python
from pyspark.sql import SparkSession
# create SparkSession
spark = SparkSession.builder \
.appName("Data Processing App") \
.getOrCreate()
# set MySQL connection properties
mysql_url = "jdbc:mysql://mysql_container:3306/sales_data"
connection_properties = {
"user": "root",
"password": "password"
}
# read data from MySQL database into DataFrame
df = spark.read.jdbc(url=mysql_url, table="sales_table", properties=connection_properties)
# perform data processing tasks
# ...
# write output data to CSV files
df1.write.csv("total_sale_per_month.csv")
df2.write.csv("ranked_items_per_month.csv")In this modified code, we have used the jdbc() method of SparkSession to read data from MySQL database. The mysql_url variable contains the JDBC URL for MySQL, and the connection_properties dictionary contains the username and password for MySQL. We don't need to specify the JDBC driver as it is already included in Spark's built-in MySQL connector.
After reading data into DataFrame, we can perform data processing tasks as required and write the output data to CSV files.
Modify the above code such that it uses mysql.connector to connect with the MySQL database.
Here's an example of how to modify the code to use mysql.connector instead of jdbc:
python
import mysql.connector
from pyspark.sql import SparkSession
# create connection to MySQL database
db_conn = mysql.connector.connect(
host="localhost",
user="root",
password="password",
database="sales_data"
)
# create SparkSession
spark = SparkSession.builder.appName("sales_data_analysis").getOrCreate()
# create DataFrame from MySQL table
df = spark.read.format("jdbc") \
.option("url", "jdbc:mysql://localhost:3306/sales_data") \
.option("user", "root") \
.option("password", "password") \
.option("dbtable", "sales_table") \
.load()
# perform data processing tasks on the DataFrame
# ...
# write the results to CSV files
# ...
# stop the SparkSession
spark.stop()
# close the database connection
db_conn.close()Note that this code assumes that the MySQL server is running on the same machine as the Spark application, and that the mysql-connector-python package is installed in the Python environment used by the Spark application. If the MySQL server is running on a different machine, you will need to replace "localhost" with the IP address or hostname of the machine running the MySQL server.
Modify the above code such that MySQL database is imported to spark dataframe without using JDBC driver.
Here's the modified code to import data from MySQL database to PySpark DataFrame without using JDBC driver:
python
import mysql.connector
from pyspark.sql import SparkSession
# MySQL database configuration
mysql_host = "localhost"
mysql_user = "root"
mysql_password = "password"
mysql_db = "sales_data"
mysql_table = "sales"
# Create a MySQL connection
cnx = mysql.connector.connect(user=mysql_user, password=mysql_password,
host=mysql_host, database=mysql_db)
# Read data from MySQL table into a pandas dataframe
query = f"SELECT * FROM {mysql_table}"
data = pd.read_sql(query, con=cnx)
# Convert pandas dataframe to PySpark dataframe
spark = SparkSession.builder.appName("SalesAnalysis").getOrCreate()
df = spark.createDataFrame(data)
# Data processing tasks on PySpark dataframe
# ...In this modified code, we use the mysql.connector library to create a MySQL connection and fetch data from the database using a SQL query. Then we convert the resulting pandas dataframe to a PySpark dataframe using the createDataFrame() method of SparkSession. We can then perform our data processing tasks on the PySpark dataframe.
Modify the below code such that environment variables such as MYSQL_DATABASE and MYSQL_ROOT_PASSWORD are set: services: mysql: build: ./mysql/. ports: - 3306:3306
services: mysql: build: ./mysql/. ports: - 3306:3306 environment: MYSQL_DATABASE: sales_data MYSQL_ROOT_PASSWORD: password
Modify the below code such that it excludes the volume part and instead includes the "command" which needs to be executed in that container. data_processing: build: ./data_processing/. volumes: - ./output:/output - ./html:/html depends_on: - mysql
Here's the modified code that excludes the volume part and includes the "command":
data_processing: build: ./data_processing/. depends_on: - mysql command: ["python", "/app/data_processing_script.py"]
Modify the above code such that the container runs indefinitely and does not die immediately.
To modify the above code to make the container run indefinitely, add the "command" parameter with the value of "tail -f /dev/null". This will prevent the container from exiting immediately after startup. Here's the modified code:
data_processing: build: ./data_processing/. volumes: - ./output:/output - ./html:/html depends_on: - mysql command: tail -f /dev/null
How to write both ["python3", "/app/data_processing_script.py"] and tail -f /dev/null in the command section?
To run both "python3 /app/data_processing_script.py" and "tail -f /dev/null" in a Docker container, you can separate the two commands with the "&&" operator in the command section of the Dockerfile. Here's an example:
bash
command: bash -c "python3 /app/data_processing_script.py && tail -f /dev/null"This will first run the Python script and then run the tail command, which will keep the container running indefinitely.
Modify the below service of yaml file such that, the container does not die immediately. data_processing: build: ./data_processing/. command: bash -c "python3 /app/data_processing_script.py && tail -f /dev/null" depends_on: - mysql
The service can be modified as follows to keep the container running indefinitely:
yaml
data_processing:
build: ./data_processing/.
command: sh -c "python3 /app/data_processing_script.py && tail -f /dev/null"
depends_on:
- mysqlHere, the bash command is replaced with sh command and the && operator is used to execute two commands in a single sh command. The tail -f /dev/null command will keep the container running indefinitely.
Modify "command: sh -c "python3 /app/data_processing_script.py && tail -f /dev/null"" command such that it executed indefinitely, thus enabling to docker container to be alive forever.
To modify the command to keep the container running indefinitely, you can replace the tail -f /dev/null command with a simple sleep infinity command as shown below:
bash
command: sh -c "python3 /app/data_processing_script.py && sleep infinity"This will execute the python3 /app/data_processing_script.py command and then keep the container running indefinitely, until it is manually stopped.
Modify the above command such that the container does not die immediately.
To modify the command such that the container does not die immediately, we can replace the "tail -f /dev/null" with "sleep infinity". The updated command would be:
bash
command: sh -c "python3 /app/data_processing_script.py && sleep infinity"This will keep the container running indefinitely until it is manually stopped.
The container is dying immediately even after the use of the above command. Please modify it.
If the container is still dying immediately, it could be due to an error in the script or some other issue. However, here is another possible modification to try:
bash
command: bash -c "python3 /app/data_processing_script.py & tail -f /dev/null"This command runs the data_processing_script.py in the background using the & operator, and then tails the /dev/null file to keep the container alive.
Identify the problem in below script which is leading to some error and therefore the container for this application could not be build. import pyspark.sql.functions as F import mysql.connector from pyspark.sql import SparkSession # MySQL database configuration mysql_host = "localhost" mysql_user = "root" mysql_password = "password" mysql_db = "sales_data" mysql_table = "sales" # Create a MySQL connection cnx = mysql.connector.connect(user=mysql_user, password=mysql_password, host=mysql_host, database=mysql_db) # Read data from MySQL table into a pandas dataframe query = f"SELECT * FROM {mysql_table}" data = pd.read_sql(query, con=cnx) # Convert pandas dataframe to PySpark dataframe spark = SparkSession.builder.appName("SalesAnalysis").getOrCreate() df = spark.createDataFrame(data) # calculate total sale for each month monthly_sales_df = sales_df \ .groupBy(F.month("order_date").alias("month"), F.year("order_date").alias("year")) \ .agg(F.sum("quantity_ordered" * "price_each").alias("total_sales")) \ .orderBy(F.desc("total_sales")) # rank items based on total units sold each month monthly_item_sales_df = sales_df \ .groupBy(F.month("order_date").alias("month"), F.year("order_date").alias("year"), "product_name") \ .agg(F.sum("quantity_ordered").alias("total_units_sold")) \ .orderBy(F.desc("total_units_sold")) # write output to csv files monthly_sales_df.write.mode("overwrite").csv("monthly_sales.csv") monthly_item_sales_df.write.mode("overwrite").csv("monthly_item_sales.csv")
The issue with the script is that it is using pandas to read data from MySQL table, but the script is missing the import statement for pandas. Additionally, the script is using sales_df to group by and create monthly sales and monthly item sales dataframes, but it should use df which is the PySpark dataframe created from the pandas dataframe.
Here's the modified code:
makefile
import pyspark.sql.functions as F
import mysql.connector
import pandas as pd
from pyspark.sql import SparkSession
# MySQL database configuration
mysql_host = "localhost"
mysql_user = "root"
mysql_password = "password"
mysql_db = "sales_data"
mysql_table = "sales"
# Create a MySQL connection
cnx = mysql.connector.connect(user=mysql_user, password=mysql_password,
host=mysql_host, database=mysql_db)
# Read data from MySQL table into a pandas dataframe
query = f"SELECT * FROM {mysql_table}"
data = pd.read_sql(query, con=cnx)
# Convert pandas dataframe to PySpark dataframe
spark = SparkSession.builder.appName("SalesAnalysis").getOrCreate()
df = spark.createDataFrame(data)
# calculate total sale for each month
monthly_sales_df = df \
.groupBy(F.month("order_date").alias("month"), F.year("order_date").alias("year")) \
.agg(F.sum(F.col("quantity_ordered") * F.col("price_each")).alias("total_sales")) \
.orderBy(F.desc("total_sales"))
# rank items based on total units sold each month
monthly_item_sales_df = df \
.groupBy(F.month("order_date").alias("month"), F.year("order_date").alias("year"), "product_name") \
.agg(F.sum("quantity_ordered").alias("total_units_sold")) \
.orderBy(F.desc("total_units_sold"))
# write output to csv files
monthly_sales_df.write.mode("overwrite").csv("monthly_sales.csv")
monthly_item_sales_df.write.mode("overwrite").csv("monthly_item_sales.csv")Make sure to install pandas module in your environment.
Modify the below portion of the above code such that mysql server defined in the yaml file is used. # MySQL database configuration mysql_host = "localhost" mysql_user = "root" mysql_password = "password" mysql_db = "sales_data" mysql_table = "sales"
To use the MySQL server defined in the docker-compose.yaml file, we can modify the MySQL database configuration as follows:
makefile
# MySQL database configuration
mysql_host = "mysql"
mysql_user = "root"
mysql_password = "password"
mysql_db = "sales_data"
mysql_table = "sales"Here, we are using the mysql service name defined in the docker-compose.yaml file as the mysql_host value, as this is the network address that the Python script container will use to communicate with the MySQL server container.
Write requirements.txt file which list all the libraries required for functioning of the application. Use pip to install then in the Dockerfile for that container.
To list all the required libraries, you can use the pip freeze command in your local environment. This command generates a list of all installed packages and their versions. You can redirect this output to a requirements.txt file using the > symbol. Here is an example:
pip freeze > requirements.txt
This will create a requirements.txt file in the current directory containing all the installed packages and their versions.
In your Dockerfile, you can use the RUN command to install the required packages from the requirements.txt file. Here is an example:
sql
FROM python:3.8
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD [ "python", "./app.py" ]This Dockerfile will first copy the requirements.txt file to the container's /app directory and then install all the packages listed in the file using pip. The --no-cache-dir flag is used to prevent the build cache from being used when installing packages, which ensures that the latest versions of the packages are installed.
After installing the packages, the Dockerfile copies the rest of the application code to the container and sets the CMD instruction to run the app.py script.
Write the contents of requirements.txt file.
Sure, here's an example of what the contents of a requirements.txt file might look like:
makefile
pyspark==3.2.1
mysql-connector-python==8.0.27
pandas==1.3.4This file includes the names and versions of three Python libraries that the application depends on: pyspark, mysql-connector-python, and pandas. You can include additional libraries in the file, each on a separate line. To install these libraries using pip in the Dockerfile, you would add the following line:
bash
RUN pip install -r /path/to/requirements.txtwhere /path/to/requirements.txt is the path to the requirements.txt file within the Docker container.