Automating Job Market Data Collection: API Calls, Data Cleaning, and Storage

import pandas as pd
import argparse
import os


def sub_object_json_extractor(df,json_col,columns_to_keep):
    loce_df = df[['PositionID',json_col]].explode(json_col)#this wil be moved to its own table

    # Flatten the nested dictionary in the 'PositionLocation' column
    location_df = pd.json_normalize(loce_df[json_col])
    # Merge back into the original dataframe
    locations = loce_df.join(location_df[columns_to_keep]).drop(json_col,axis=1)
    return locations

def remuneration_extractor(df):
    # Flatten the nested dictionary in the 'PositionLocation' column
    remuneration_df_explode = df.explode('PositionRemuneration')
    remuneration_df = pd.json_normalize(remuneration_df_explode['PositionRemuneration'])

    # Select only the required columns
    columns_to_keep = ['MinimumRange', 'MaximumRange','RateIntervalCode']

    # Merge back into the original dataframe
    df = df.join(remuneration_df[columns_to_keep])

    df['MinimumRange'] = pd.to_numeric(df['MinimumRange'])
    df['MaximumRange'] = pd.to_numeric(df['MaximumRange'])
    #feature generating
    #Yearly Salary=Hourly Wage×40×52
    def normalized_salary(row,col_name):
        s = row[col_name]
        rateCode = row['RateIntervalCode']
        if rateCode=='PH':
            return 40*52*s
        elif rateCode=='PA':
            return s

    df['NormalizedMinSalary'] = df.apply(lambda row: normalized_salary(row, 'MinimumRange'), axis=1)
    df['NormalizedMaxSalary'] = df.apply(lambda row: normalized_salary(row, 'MaximumRange'), axis=1)
    df['NormalizedMeanSalary'] = df[["NormalizedMinSalary", "NormalizedMaxSalary"]].mean(axis=1)
    df['NormalizedRangeSalary'] = df['NormalizedMaxSalary']-df['NormalizedMinSalary']
    return df


def write_to_parquet(jobs,duties,locations,job_category,first_run):
    os.makedirs('./job-market-gov/data/', exist_ok=True)
    # Path to the Parquet file
    files_dict = {'jobs':jobs,
                    'duties':duties,
                    'locations':locations,
                    'job_category':job_category}
    for df_name,df_current in files_dict.items():
        p_path = f'./job-market-gov/data/{df_name}.parquet'
        if first_run:
            # Write the DataFrame back to the Parquet file
            df_current.to_parquet(p_path, engine='pyarrow', index=False)
        else:
            try:
                # Try reading the existing Parquet file
                existing_df = pd.read_parquet(p_path)
                # Append the new data to the existing DataFrame
                combined_df = pd.concat([existing_df, df_current], ignore_index=True)
            except FileNotFoundError:
                # If the file doesn't exist, create a new DataFrame with the new data
                combined_df = df_current

            # Write the combined DataFrame back to the Parquet file
            combined_df.to_parquet(p_path, engine='pyarrow', index=False)


def main(path,first_run):
    print('starting')
    df_clean = pd.read_json(path)
    userArea = pd.json_normalize(df_clean['UserArea'],sep='_')
    df_clean = df_clean.join(userArea)
    locations = sub_object_json_extractor(df=df_clean,json_col='PositionLocation',columns_to_keep=['LocationName', 'CountryCode', 'CountrySubDivisionCode', 'CityName', 'Longitude', 'Latitude'])
    job_category = sub_object_json_extractor(df=df_clean,json_col='JobCategory',columns_to_keep=['Name', 'Code'])
    df_clean = remuneration_extractor(df=df_clean)
    df_clean['date_col'] = pd.to_datetime(df_clean['PublicationStartDate']).dt.date
    duties = df_clean[['PositionID','Details_MajorDuties']].explode('Details_MajorDuties')
    jobs = df_clean.drop(['UserArea','PositionLocation','JobCategory','Details_MajorDuties','MinimumRange','MaximumRange','RateIntervalCode','PublicationStartDate','PositionRemuneration'],axis=1)
    write_to_parquet(jobs,duties,locations,job_category,first_run)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Process job market data and store in parquet files.")
    parser.add_argument("path", type=str, help="Path to the JSON file")
    parser.add_argument("--first_run",
                        type=bool,
                        default=False,  # Default value if not provided
                        help="Indicate if it's the first run or not (default is False)")
    
    args = parser.parse_args()
    # Access the value of first_run
    main(args.path,args.first_run)
                            
                        

{
    "PositionID": "DE-11724537-25-MRT",
    "PositionTitle": "CIVIL ENGINEER (STRUCTURAL)",
    "PositionLocationDisplay": "Naval Base Newport, Rhode Island",
    "PositionLocation": [
    {
        "LocationName": "Naval Base Newport, Rhode Island",
        "CountryCode": "United States",
        "CountrySubDivisionCode": "Rhode Island",
        "CityName": "Naval Base Newport, Rhode Island",
        "Longitude": -71.41487,
        "Latitude": 41.595963
    }
    ],
    "OrganizationName": "Naval Facilities Engineering Systems Command",
    "DepartmentName": "Department of the Navy",
    "JobCategory": [
    {
        "Name": "Civil Engineering",
        "Code": "0810"
    }
    ],
    "QualificationSummary": "To qualify for the GS-12: In addition to the Basic Requirements for this position, your resume must also demonstrate at least one year of specialized experience as a professional civil engineer at or equivalent to the GS-11 grade level or pay band in the Federal service or equivalent experience in the private or public sector. Specialized experience must demonstrate the 1) Using advanced concepts, principles, and practices in engineering/architecture to enable incumbent to serve as an expert in the complex multiple engineering areas of consultation, design, and construction; 2) Performing structural project development and design; 3) Preparing project specifications and cost estimates; 4) Utilizing computers for design, drafting and preparation on calculations and structural engineering drawings. Additional qualification information can be found from the following Office of Personnel Management website: https://www.opm.gov/policy-data-oversight/classification-qualifications/general-schedule-qualification-standards/#url=GS-PROF AND https://www.opm.gov/policy-data-oversight/classification-qualifications/general-schedule-qualification-standards/0800/files/all-professional-engineering-positions-0800.pdf Experience refers to paid and unpaid experience, including volunteer work done through National Service programs (e.g., professional, philanthropic, religious, spiritual, community, student, social). Volunteer work helps build critical competencies, knowledge, and skills and can provide valuable training and experience that translates directly to paid employment.",
    "PositionRemuneration": [
    {
        "MinimumRange": "100371.0",
        "MaximumRange": "130488.0",
        "RateIntervalCode": "PA",
        "Description": "Per Year"
    }
    ],
    "PublicationStartDate": "2025-01-16T00:00:00.0000",
    "UserArea": {
    "Details": {
        "JobSummary": "You will serve as a Civil Engineer (Structural) in the Capital Improvements Division of NAVFAC ENGINEERING CMD MID-ATLANTIC.",
        "MajorDuties": [
        "You will assess structural condition of facilities to determine needs for repairs and modifications.",
        "You will analyze and design extensive structural alterations.",
        "You will perform quality assurance review of drawings, specifications, calculations, cost estimates, and studies prepared by A/E design firms.",
        "You will create written documents such as specifications, test requirement documents, and drawings for civil engineering projects.",
        "You will complete design drawings and specifications with design and drafting software.",
        "You will review plans, designs, and construction phases to ensure civil engineering projects remain in scope.",
        "You will investigate construction to make recommendations for remediation or corrective measures.",
        "You will advise on solutions to problems encountered during projects."
        ]
    }
    }
}
                        

FROM ubuntu:latest

COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

WORKDIR /app

RUN apt-get update && apt-get install -y curl \
                                            jq \
                                            python3.12 \
                                            && rm -rf /var/lib/apt/lists/*


# Copy the project into the image
COPY pyproject.toml run.sh uv.lock job-market-gov/job_listing.py .

# Define build arguments
ARG API_EMAIL
ARG API_AUTH

# Set environment variables from build arguments
ENV API_EMAIL=${API_EMAIL}
ENV API_AUTH=${API_AUTH}

# Give execute permissions to the shell script
RUN chmod +x ./run.sh && ./run.sh

# Create a virtual environment and install dependencies
RUN uv venv && \
    . .venv/bin/activate && \
    uv sync --frozen && \
    python3 job_listing.py main.json --first_run True