sfleads_transform.py

import pandas as pd
import json

df = pd.read_csv(
    's3://airetail/bronze/sfleads/addresses/Addresses.csv'
)

df.address = df['address'].apply(json.loads)
df.phone_numbers = df['phone_numbers']. \
    apply(lambda pn: json.loads(pn) if not pd.isnull(pn) else None)

df_normalized = pd.json_normalize(df.to_dict(orient='records'))
df_normalized.columns = [
    col.split('.')[1] if len(col.split('.')) == 2 else col for col in df_normalized.columns
]

df_normalized.to_csv(
    's3://airetail/gold/sfleads/addresses/Addresses.csv', 
    index=None
)

sfleads_transform_lambda.py

import json
import os
import boto3
import pandas as pd


def lambda_handler(event, context):
    bucket_name = os.environ.get('BUCKET_NAME')
    s3_client = boto3.client('s3')
    df = pd.read_csv(
        f's3://{bucket_name}/bronze/addresses/Addresses.csv'
    )
    df.address = df['address'].apply(json.loads)
    df.phone_numbers = df['phone_numbers']. \
        apply(lambda pn: json.loads(pn) if not pd.isnull(pn) else None)
    df_exploded = df.explode('phone_numbers').rename(columns={'phone_numbers': 'phone_number'})
    df_normalized = pd.json_normalize(df_exploded.to_dict(orient='records'))
    df_normalized.columns = [
        col.split('.')[1] if len(col.split('.')) == 2 else col for col in df_normalized.columns
    ]
    df_normalized.to_csv(
        f's3://{bucket_name}/gold/addresses/Addresses.csv', 
        index=None
    )
    return {
        'statusCode': 200,
        'statusMessage': 'Successfully Transformed the data'
    }