Standard template Data Engineering and Analytics - hosthub

[Hosthubac]

Data Engineering Template (ETL: S3 → EMR/PySpark → S3 or Redshift)

# Imports
import boto3
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, lit
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Initialize Spark Session
spark = SparkSession.builder \
    .appName("ETL Pipeline") \
    .getOrCreate()

# Read Data from S3
def read_s3_data(bucket_name, key, schema=None, file_format="csv"):
    s3_path = f"s3://{bucket_name}/{key}"
    if file_format == "csv":
        return spark.read.option("header", True).schema(schema).csv(s3_path)
    elif file_format == "parquet":
        return spark.read.parquet(s3_path)
    else:
        raise ValueError("Unsupported file format")

# Example schema (optional)
example_schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("amount", IntegerType(), True)
])

# df = read_s3_data("my-bucket", "data/input.csv", example_schema)
# Transform Data
def transform_data(df):
    df_transformed = df.withColumn("amount_double", col("amount")*2) \
                       .withColumn("category", when(col("amount")>100, "High").otherwise("Low"))
    return df_transformed

# df_transformed = transform_data(df)
# Load Data to S3
def write_s3_data(df, bucket_name, key, file_format="parquet"):
    s3_path = f"s3://{bucket_name}/{key}"
    if file_format == "csv":
        df.write.mode("overwrite").option("header", True).csv(s3_path)
    elif file_format == "parquet":
        df.write.mode("overwrite").parquet(s3_path)
    else:
        raise ValueError("Unsupported file format")

# write_s3_data(df_transformed, "my-bucket", "data/output/")
# Load Data to Redshift
def write_redshift(df, jdbc_url, table_name, user, password):
    df.write \
      .format("jdbc") \
      .option("url", jdbc_url) \
      .option("dbtable", table_name) \
      .option("user", user) \
      .option("password", password) \
      .mode("append") \
      .save()

Notes:
Use def functions for modularity.
Easy to replace S3 paths, schema, and transformations.
Redshift loading uses Spark JDBC.

Data Analytics Template (Boto3, SageMaker, Sklearn, Pandas, S3)

# Imports
import boto3
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error
import joblib

# Read Data from S3
s3_client = boto3.client('s3')
bucket_name = "my-bucket"
key = "data/input.csv"
obj = s3_client.get_object(Bucket=bucket_name, Key=key)
df = pd.read_csv(obj['Body'])

# Basic EDA / Analytics
print(df.head())
print(df.describe())
print(df.info())

# Train/Test Split
X = df.drop(columns=['target'])
y = df['target']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Train Model (Sklearn)
model = LinearRegression()
model.fit(X_train, y_train)

# Evaluate Model
y_pred = model.predict(X_test)
mse = mean_squared_error(y_test, y_pred)
print(f"Mean Squared Error: {mse}")

# Save Model to S3
joblib.dump(model, "/tmp/model.joblib")
s3_client.upload_file("/tmp/model.joblib", bucket_name, "models/model.joblib")

# SageMaker (Optional for Advanced)
# Example: Create SageMaker session and upload data
import sagemaker
sagemaker_session = sagemaker.Session()
role = "arn:aws:iam::<account_id>:role/SageMakerRole"

# Upload data to S3 for SageMaker training
train_s3_path = sagemaker_session.upload_data(path="data/input.csv", key_prefix="training_data")

Notes:
Uses simple variables and Boto3 directly—no need for functions unless you want.
Covers the typical workflow: S3 → Pandas → ML → S3.
Can be extended for SageMaker training jobs.