Build a Data Analytics Pipeline with Python, SQL and Power BI (CSV to Dashboard)

A data analytics pipeline with Python, SQL and Power BI works in three hand-offs: Python (pandas) cleans the raw file, a SQL database such as MySQL stores the clean rows and does the heavy aggregation, and Power BI connects to that database to show the results as a dashboard that refreshes itself. Once it is built, nobody has to open the CSV by hand again.

This post walks through that end-to-end analytics pipeline using a small sales dataset, with working code for each stage. It also covers the parts most tutorials skip: duplicate loads, a schema that breaks on real orders, and what Power BI actually needs before scheduled refresh will work.

What does the pipeline look like?

sales_data.csv  →  Python (clean + validate)  →  MySQL (store + aggregate)  →  Power BI (dashboard)

Each tool has one job:

Stage	Tool	Job
Clean	Python + pandas	Remove duplicates, fix types, reject bad rows, recalculate revenue
Store	MySQL	Keep one trusted copy of the data, block duplicate loads, run aggregations
Show	Power BI	KPIs, trends and filters for people who will never write a query

You can swap MySQL for PostgreSQL or SQL Server. The logic stays the same; only the connection string and a few SQL functions change.

Step 1: How do you clean raw sales data with Python?

Assume the export has these columns: Order_ID, Order_Date, Customer_Name, Product, Category, Quantity, Unit_Price.

Install the three packages you need:

pip install pandas sqlalchemy pymysql

Then clean the file:

import pandas as pd

df = pd.read_csv("sales_data.csv")
print(df.shape)
print(df.isnull().sum())

# Exact duplicate rows (the same export pulled twice, for example)
df = df.drop_duplicates()

# Rows we can't use without an order ID or a product
df = df.dropna(subset=["Order_ID", "Product"])

# "Wireless Mouse " and "wireless mouse" should count as one product
df["Product"] = df["Product"].str.strip().str.title()
df["Category"] = df["Category"].str.strip().str.title()

# Dates: set the format your export really uses (e.g. "%d-%m-%Y").
# errors="coerce" turns bad values into NaT instead of crashing.
df["Order_Date"] = pd.to_datetime(df["Order_Date"], format="%Y-%m-%d", errors="coerce")
df = df.dropna(subset=["Order_Date"])

# Numbers: reject text, zero and negative values
df["Quantity"] = pd.to_numeric(df["Quantity"], errors="coerce")
df["Unit_Price"] = pd.to_numeric(df["Unit_Price"], errors="coerce")
df = df[(df["Quantity"] > 0) & (df["Unit_Price"] > 0)]

# Recalculate revenue instead of trusting a column someone typed into Excel
df["Revenue"] = (df["Quantity"] * df["Unit_Price"]).round(2)

print(f"{len(df)} clean rows ready to load")

Two lines here deserve a second look.

The date format. Many exports from Indian billing tools write dates as DD-MM-YYYY. Leave the format out and pandas may guess, so 03-04-2026 can quietly become 4 March instead of 3 April. Your monthly chart will look plausible and still be wrong. Always set format= explicitly.

.str.title() fixes casing, but it also turns "iPhone" into "Iphone". If your product names include brands like that, map them with a small dictionary instead.

Why does data cleaning matter so much?

Because every later stage trusts it. Duplicate transactions double-count revenue, broken dates fall out of monthly totals, and "Laptop" and "laptop " show up as two products in your top-5 chart. Power BI will happily draw a polished chart from bad data. Cleaning is where you stop that.

Step 2: How do you store the cleaned data in SQL without duplicates?

A common first draft uses Order_ID as the primary key. It breaks on the first real order with two products in it, because both lines share one order ID. Give each row its own ID, and put a unique key on the combination that really identifies a sale line:

CREATE DATABASE analytics_db;
USE analytics_db;

CREATE TABLE sales (
    Sale_ID        INT AUTO_INCREMENT PRIMARY KEY,
    Order_ID       INT NOT NULL,
    Customer_Name  VARCHAR(100),
    Product        VARCHAR(100) NOT NULL,
    Category       VARCHAR(50),
    Order_Date     DATE NOT NULL,
    Quantity       INT NOT NULL,
    Unit_Price     DECIMAL(10,2) NOT NULL,
    Revenue        DECIMAL(12,2) NOT NULL,
    UNIQUE KEY uq_order_product (Order_ID, Product),
    INDEX idx_order_date (Order_Date)
);

The index on Order_Date keeps date-range queries fast once the table grows past a few hundred thousand rows.

Now load from Python. Credentials come from environment variables, never from the script, and rows go into a staging table first so a re-run doesn't crash or double-load:

import os
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL

url = URL.create(
    "mysql+pymysql",
    username=os.environ["DB_USER"],
    password=os.environ["DB_PASSWORD"],
    host=os.environ.get("DB_HOST", "localhost"),
    database="analytics_db",
)
engine = create_engine(url)

cols = ["Order_ID", "Customer_Name", "Product", "Category",
        "Order_Date", "Quantity", "Unit_Price", "Revenue"]
col_list = ", ".join(cols)

with engine.begin() as conn:
    df[cols].to_sql("sales_staging", conn, if_exists="replace", index=False)
    result = conn.execute(text(
        f"INSERT IGNORE INTO sales ({col_list}) SELECT {col_list} FROM sales_staging"
    ))
    print(f"{result.rowcount} new rows loaded")

INSERT IGNORE skips any row that would break the unique key, so loading yesterday's file twice adds nothing. If corrected rows should overwrite old ones, use INSERT ... ON DUPLICATE KEY UPDATE instead.

URL.create() also saves you from a classic bug: a password containing @ or # breaks a hand-built connection string.

Step 3: Which SQL queries turn rows into business metrics?

Let the database do the aggregation. It is faster than pulling every row into Power BI and summing there.

Monthly revenue and orders

SELECT
    DATE_FORMAT(Order_Date, '%Y-%m') AS Month,
    SUM(Revenue)                     AS Total_Revenue,
    COUNT(DISTINCT Order_ID)         AS Orders
FROM sales
GROUP BY DATE_FORMAT(Order_Date, '%Y-%m')
ORDER BY Month;

Top 5 products by revenue

SELECT
    Product,
    SUM(Quantity) AS Units_Sold,
    SUM(Revenue)  AS Total_Revenue
FROM sales
GROUP BY Product
ORDER BY Total_Revenue DESC
LIMIT 5;

Headline KPIs

SELECT
    SUM(Revenue)                                   AS Total_Revenue,
    COUNT(DISTINCT Order_ID)                       AS Total_Orders,
    ROUND(SUM(Revenue) / COUNT(DISTINCT Order_ID), 2) AS Avg_Order_Value
FROM sales;

Note COUNT(DISTINCT Order_ID), not COUNT(*). With one row per product line, COUNT(*) counts lines, not orders, and your order count comes out inflated.

If a query becomes a regular report, save it as a view (CREATE VIEW v_monthly_revenue AS ...). Power BI lists views alongside tables, so your report reads from one agreed definition.

Step 4: How do you connect MySQL to Power BI?

One prerequisite catches almost everyone: Power BI Desktop needs the Oracle MySQL Connector/NET installed before the MySQL connector will work. Install it first, then restart Power BI.

1. Open Power BI Desktop and select Get data → MySQL database.
2. Enter the server (for example localhost) and the database name analytics_db.
3. Optionally open Advanced options and paste one of the SQL queries above.
4. Sign in with a read-only database user, not the account your Python script uses.
5. In the Navigator, tick the sales table and any views.
6. Click Transform data and check the column types: Order_Date should be Date, Revenue should be a decimal number.
7. Click Close & Apply.

What should the dashboard show?

Define the core measures once in DAX so every visual uses the same logic:

Total Revenue = SUM ( sales[Revenue] )
Total Orders = DISTINCTCOUNT ( sales[Order_ID] )
Avg Order Value = DIVIDE ( [Total Revenue], [Total Orders] )

Then build:

• Cards for Total Revenue, Total Orders and Avg Order Value
• A line chart of revenue by month
• A bar chart of the top products
• A bar chart of revenue by category (easier to compare than a pie once you pass four or five categories)
• Slicers for date range and category

Stop there. One page that answers "how are sales doing, and what is driving it?" beats five pages nobody opens. If people keep asking a question the page can't answer, add a visual for it then.

Step 5: How do you automate the pipeline?

Turn Steps 1 and 2 into one script (load_sales.py), add logging, and let the operating system run it on a schedule.

import logging

logging.basicConfig(
    filename="pipeline.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

try:
    # ... cleaning and loading code from Steps 1 and 2 ...
    logging.info("Loaded %s new rows", result.rowcount)
except Exception:
    logging.exception("Pipeline failed")
    raise

On Linux, a cron entry runs it every morning at 6:00:

0 6 * * * /usr/bin/python3 /opt/pipeline/load_sales.py

On Windows, Task Scheduler does the same job. Neither is a Python library; they are operating-system schedulers that run your script. Once you have several dependent jobs that need retries and alerts, a workflow tool like Apache Airflow becomes worth its setup cost. For a single daily load, cron is enough.

The Power BI side of automation happens in the Power BI service, not Desktop:

1. Publish the report from Power BI Desktop.
2. If MySQL runs on your own machine or office server, install an on-premises data gateway (standard mode; personal mode isn't supported for MySQL) and install Connector/NET on that gateway machine too. A cloud-hosted MySQL server that Power BI can reach doesn't need a gateway.
3. Set a scheduled refresh for a time after your Python job usually finishes, for example 6:30 if the script runs at 6:00. How many refreshes a day you get depends on your Power BI license.

Where do these pipelines usually break?

• The same file loaded twice. Without a unique key, revenue doubles and nobody notices until finance asks. The staging table and unique key from Step 2 prevent it.
• Silent date errors. Day and month swapped, or blanks turned into 1970-01-01. Check the min and max Order_Date after every load.
• Three definitions of "revenue". Python, SQL and Power BI should follow one rule (with or without tax? before or after discount?). Write it down and compute it in one place.
• Hardcoded passwords. They end up in Git, in screenshots and in blog posts. Use environment variables or a secrets manager.
• A refresh that fails quietly. Turn on refresh-failure email notifications in the Power BI service and read your pipeline.log.

Quick answers

Do I need Apache Airflow for this?
No. For one daily job, cron or Task Scheduler is simpler and easier to debug. Airflow pays off when you have many dependent tasks.

Can I use PostgreSQL instead of MySQL?
Yes. Change the SQLAlchemy driver (for example postgresql+psycopg2), use PostgreSQL's date functions in your queries, and pick the PostgreSQL connector in Power BI.

Is Power BI free?
Power BI Desktop is a free download, so you can build this whole project at no cost. Publishing and sharing reports with other people in the Power BI service generally needs a paid license.

Is this a good portfolio project?
It is a solid one, because it shows the full workflow rather than just a chart. Put the code on GitHub, add a screenshot of the dashboard, and explain one data-quality problem you found and how you fixed it.

What would you add?

A cloud data warehouse? dbt for the SQL layer? A forecasting model on top of the monthly revenue? Tell me in the comments how you would extend this pipeline.

---

Written by the team at Solitaire Infosystems, a software development and IT training company in Mohali, Punjab, that runs job-oriented programs in Python, ML/AI and other technologies, along with industrial training and internships.