PYTHON FOR DATA ENGINEERING
3 From Basics to Production | pandas | File Handling | GCP Python SDK | Labs
3.1 Why Python? (For Non-Programmers)
Python is the most popular language for data engineering because it reads almost like English, has thousands
of ready-made tools (libraries), and connects to every data system ever built. If SQL is how you talk to
databases, Python is how you build everything around the database.
💡 REAL-LIFE ANALOGY
If SQL is like placing a food order (asking for what you want),
Python is like being in the kitchen (cooking, preparing, delivering the food).
Python is your Swiss Army knife — it can do almost anything data-related.
3.2 Python Fundamentals (Quick Reference)
# Variables — storing values
customer_name = 'Ravi Kumar' # String (text)
order_amount = 1499.99 # Float (decimal number)
item_count = 3 # Integer (whole number)
is_premium = True # Boolean (True/False)
# Lists — ordered collection (like a column of values)
cities = ['Mumbai', 'Delhi', 'Bangalore', 'Hyderabad']
amounts = [1000, 2500, 750, 3200, 890]
# Dictionaries — key-value pairs (like a single database row)
customer = {
'id': 101,
'name': 'Ravi Kumar',
'city': 'Mumbai',
'total_orders': 12
}
print(customer['name']) # Output: Ravi Kumar
# Functions — reusable blocks of code
def calculate_discount(price, pct):
discount = price * (pct / 100)
return price - discount
final_price = calculate_discount(1000, 10) # Returns: 900.0
# Loops — repeat actions
for city in cities:
print(f'Processing orders for: {city}')
# Conditions
if order_amount > 1000:
print('Eligible for free shipping')
elif order_amount > 500:
GCP Cloud Data Engineer — Zero to Hero All Industries | Hands-On | Zero to Hero
print('Eligible for discount')
else:
print('Standard order')
3.3 pandas — Data Engineering Workhorse
pandas is the most important Python library for data engineering. It lets you work with tables of data (called
DataFrames) — reading files, cleaning data, transforming, and more.
import pandas as pd
import numpy as np
# --- Reading Data ---
# From CSV file
df = pd.read_csv('[Link]')
# From PostgreSQL database
from sqlalchemy import create_engine
engine = create_engine('postgresql://dataeng:dataeng123@localhost/sourcedb')
df = pd.read_sql('SELECT * FROM [Link]', engine)
# --- Exploring Data (always do this first!) ---
print([Link]) # (rows, columns)
print([Link]) # Data types of each column
print([Link](5)) # First 5 rows
print([Link]()) # Statistical summary
print([Link]().sum()) # Count missing values per column
# --- Filtering (like SQL WHERE) ---
completed = df[df['status'] == 'completed']
high_value = df[df['total_amount'] > 5000]
mumbai_orders = df[(df['city']=='Mumbai') & (df['status']=='completed')]
# --- Cleaning Data ---
# Fill missing values
df['city'].fillna('Unknown', inplace=True)
df['total_amount'].fillna(0, inplace=True)
# Remove duplicates
df.drop_duplicates(subset=['order_id'], inplace=True)
# Standardize text
df['city'] = df['city'].[Link]().[Link]() # 'mumbai ' → 'Mumbai'
# Convert data types
df['order_date'] = pd.to_datetime(df['order_date'])
df['total_amount'] = pd.to_numeric(df['total_amount'], errors='coerce')
# --- Transformations ---
df['month'] = df['order_date'].dt.to_period('M')
df['profit'] = df['total_amount'] - df['cost']
df['is_high_value'] = df['total_amount'] > 5000
# --- Aggregations (like SQL GROUP BY) ---
revenue_by_city = [Link]('city').agg(
total_orders=('order_id', 'count'),
total_revenue=('total_amount', 'sum'),
avg_order=('total_amount', 'mean')
).reset_index()
GCP PDE | 6 Sigma | Docker + GCP Hands-On Page 2 E-Commerce | Healthcare | Finance
GCP Cloud Data Engineer — Zero to Hero All Industries | Hands-On | Zero to Hero
# --- Writing Output ---
df.to_csv('cleaned_orders.csv', index=False)
df.to_parquet('[Link]', index=False) # Better format for pipelines
df.to_sql('orders_clean', engine, schema='analytics', if_exists='replace')
⚓ HANDS-ON LAB
LAB 3.1 — Python ETL Script: PostgreSQL → Cleaned CSV → BigQuery
Goal: Read orders from PostgreSQL, clean the data, write to GCS and BigQuery
File: etl_orders.py
Run: python3 etl_orders.py
Expected output: 'Successfully loaded 450 rows to BigQuery'
Next: Verify in BigQuery console — bq show your_project:dataset.orders_clean
# etl_orders.py — Full ETL Pipeline in Python
import pandas as pd
from sqlalchemy import create_engine
from [Link] import bigquery, storage
import logging
from datetime import datetime
[Link](level=[Link], format='%(asctime)s - %(message)s')
log = [Link](__name__)
# ■■ Config ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
PG_CONN = 'postgresql://dataeng:dataeng123@localhost/sourcedb'
GCS_BUCKET = 'your-data-lake-bucket'
BQ_TABLE = '[Link].orders_clean'
def extract(engine):
[Link]('Extracting data from PostgreSQL...')
query = '''
SELECT o.order_id, o.order_date, [Link], o.total_amount,
c.full_name AS customer_name, [Link], [Link],
p.product_name, [Link]
FROM [Link] o
JOIN [Link] c USING (customer_id)
JOIN ecommerce.order_items oi USING (order_id)
JOIN [Link] p USING (product_id)
'''
df = pd.read_sql(query, engine)
[Link](f'Extracted {len(df)} rows')
return df
def transform(df):
[Link]('Transforming data...')
df = df.drop_duplicates(subset=['order_id'])
df['order_date'] = pd.to_datetime(df['order_date'])
df['total_amount'] = pd.to_numeric(df['total_amount'], errors='coerce').fillna(0)
df['city'] = df['city'].[Link]().[Link]()
df['month'] = df['order_date'].dt.to_period('M').astype(str)
df = df[df['total_amount'] > 0] # Remove zero-amount orders
[Link](f'After transform: {len(df)} rows')
return df
GCP PDE | 6 Sigma | Docker + GCP Hands-On Page 3 E-Commerce | Healthcare | Finance
GCP Cloud Data Engineer — Zero to Hero All Industries | Hands-On | Zero to Hero
def load_to_bq(df, table_id):
client = [Link]()
job_config = [Link](
write_disposition='WRITE_TRUNCATE',
autodetect=True
)
job = client.load_table_from_dataframe(df, table_id, job_config=job_config)
[Link]()
[Link](f'Loaded {len(df)} rows to {table_id}')
if __name__ == '__main__':
engine = create_engine(PG_CONN)
df_raw = extract(engine)
df_clean = transform(df_raw)
load_to_bq(df_clean, BQ_TABLE)
print('Pipeline completed successfully!')
GCP PDE | 6 Sigma | Docker + GCP Hands-On Page 4 E-Commerce | Healthcare | Finance