insurance-usa-citizens-categories-and-dental-cost.ipynb

{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!pip install pyspark","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pyspark.sql import SparkSession\nfrom pyspark.sql.functions import *\nfrom pyspark.sql.types import *\nfrom pyspark.sql import functions as F\nimport os\nimport pandas as pd\nimport numpy as np\nimport geopandas as gpd\nimport matplotlib.pyplot as plt\nimport seaborn as sns","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Create Spark Session\nspark = SparkSession.builder \\\n    .appName(\"ExploratoryAnalysis\") \\\n    .getOrCreate()\n","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Plan Attriubutes CSV\ndf_planAttributes = spark.read.csv(\"raw_data/PlanAttributes.csv\", header=True, inferSchema=True)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Categorization of customers based on the market value\n# Categories: \n#Bronze - 60%, Silver - 70%, Gold - 80%, Platinum - 90%, Catastrophic - below 60% sold to individual","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df_IssuerValue= df_planAttributes[df_planAttributes['IssuerActuarialValue'] != '']\ndf_IssuerValue = df_planAttributes.select('StateCode', 'IssuerId', 'ServiceAreaId', 'IssuerActuarialValue', 'MarketCoverage')\ndf_IssuerValue = df_IssuerValue.withColumn(\"IssuerActuarialValue\", F.regexp_replace(F.col(\"IssuerActuarialValue\"), \"%\", \"\"))\ndf_IssuerValue = df_IssuerValue.withColumn(\"IssuerActuarialValue\", F.col(\"IssuerActuarialValue\").cast(\"int\"))\n\n#Categories Gold,Bronze,Silver,Platinum\ndf_bronze = df_IssuerValue.filter((F.col(\"IssuerActuarialValue\") >= 60) & (F.col(\"IssuerActuarialValue\") < 70))\ndf_bronze = df_bronze.groupBy(\"StateCode\").agg(round(mean(\"IssuerActuarialValue\"), 2).alias(\"Bronze\"))\n\ndf_Silver = df_IssuerValue.filter((F.col(\"IssuerActuarialValue\") >= 70) & (F.col(\"IssuerActuarialValue\") < 80))\ndf_Silver = df_Silver.groupBy(\"StateCode\").agg(round(mean(\"IssuerActuarialValue\"),2).alias(\"Silver\"))\n\ndf_Gold= df_IssuerValue.filter((F.col(\"IssuerActuarialValue\") >= 80) & (F.col(\"IssuerActuarialValue\") < 90))\ndf_Gold= df_Gold.groupBy(\"StateCode\").agg(round(mean(\"IssuerActuarialValue\"),2).alias(\"Gold\"))\n\ndf_Platinum= df_IssuerValue.filter((F.col(\"IssuerActuarialValue\") >= 90) & (F.col(\"IssuerActuarialValue\") <= 100))\ndf_Platinum= df_Platinum.groupBy(\"StateCode\").agg(round(mean(\"IssuerActuarialValue\"),2).alias(\"Platinum\"))\n\ndf_catastrophic= df_IssuerValue.filter(col(\"IssuerActuarialValue\") < 60)\ndf_catastrophic = df_catastrophic.groupBy(\"StateCode\").agg(round(mean(\"IssuerActuarialValue\"),2).alias(\"Catastrophic\"))","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Comdine all the Catergories in order to make a new dataset\nfrom functools import reduce\nfrom pyspark.sql import DataFrame\ndef join_dfs(df1, df2, join_column_name=\"StateCode\"):\n    return df1.join(df2, on=join_column_name, how=\"outer\")\n\nall_tiers = [df_bronze, df_Silver,df_Gold,df_Platinum, df_catastrophic]\ndf_categories= reduce(join_dfs, all_tiers)\n\n# Remove_List contains US regions like the Counry Colombia\nremove_list = [\n    \"30751MT0560005\", \"30751MT0560006\", \"30751MT0560014\", \"30751MT0560015\",\n    \"30751MT0560020\", \"30751MT0560021\"\n]\n\n# Filter out the rows with the values in the remove_list from the StateCode column\ndf_categories = df_categories.filter(~col(\"StateCode\").isin(remove_list))\ndf_categories.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Graphic Map of US","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Join categories with map \ndf_usGeo=spark.read.csv(\"geo_data/US_GeoCode.csv\", header=True, inferSchema=True)\n\n# Rename the 'StateCode' column in df_usGeo to 'region'\ndf_usGeo = df_usGeo.withColumnRenamed(\"StateCode\", \"region\")\n# Rename the 'StateCode' column in df_categories to 'region'\ndf_categories = df_categories.withColumnRenamed(\"StateCode\", \"region\")\n\n# Group by 'region' and compute average longitude and latitude\nstatename = df_usGeo.groupBy(\"region\").agg(avg(\"longitude\").alias(\"long\"), avg(\"latitude\").alias(\"lat\"))\n\n# Join the two DataFrames on the 'region' column\nmapdata = df_categories.join(df_usGeo, on=\"region\", how=\"left_outer\")\nmapdata_pd = mapdata.toPandas()\n\n# Loading data and the US states shapefile\nus_states_detailed = gpd.read_file(\"geo_data/tl_2022_us_state.shp\")\nmapdata_gdf = gpd.GeoDataFrame(mapdata_pd, geometry=gpd.points_from_xy(mapdata_pd.longitude, mapdata_pd.latitude))\n\n# Ensure the CRS matches before spatial join\n# Set the initial CRS for mapdata_gdf\nmapdata_gdf.crs = \"EPSG:4326\"\n\n#Transformation in order to match the US states shapefile's CRS\nmapdata_gdf = mapdata_gdf.to_crs(us_states_detailed.crs)\n\n\n# Join the data using a spatial join\nmerged_data = gpd.sjoin(us_states_detailed, mapdata_gdf, how=\"left\", predicate=\"intersects\")\n","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Catastrophic Map\nfig, ax = plt.subplots(figsize=(20, 15))  # Adjust the size here for a bigger map\n\nmerged_data.boundary.plot(ax=ax, color='black', linewidth=0.5)\nmerged_data.plot(column='Catastrophic', ax=ax, legend=True, cmap='Reds', legend_kwds={'label': \"Catastrophic Coverage\"})\n\n# Bounds to focus on the contiguous US\nax.set_xlim(-130, -60)\nax.set_ylim(20, 50)\n\n# Centering the title\nax.set_title(\"Catastrophic Coverage in the US\", fontsize=20, pad=20)\n\n# Removing the axis for a cleaner look\nax.axis('off')\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Platinum Map\nfig, ax = plt.subplots(figsize=(15, 10))  # Adjust the size here for a bigger map\n\nmerged_data.boundary.plot(ax=ax, color='black', linewidth=0.5)\nmerged_data.plot(column='Platinum', ax=ax, legend=True, cmap='Greens', legend_kwds={'label': \"Platinum Coverage\"})\n\n# Bounds to focus on the contiguous US\nax.set_xlim(-130, -60)\nax.set_ylim(20, 50)\n\n# Centering the title\nax.set_title(\"Platinum Coverage in the US\", fontsize=20, pad=20)\n\n# Removing the axis for a cleaner look\nax.axis('off')\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Bronze Map\nfig, ax = plt.subplots(figsize=(15, 10))  # Adjust the size here for a bigger map\n\nmerged_data.boundary.plot(ax=ax, color='black', linewidth=0.5)\nmerged_data.plot(column='Bronze', ax=ax, legend=True, cmap='Oranges', legend_kwds={'label': \"Bronze Coverage\"})\n\n# Bounds to focus on the contiguous US\nax.set_xlim(-130, -60)\nax.set_ylim(20, 50)\n\n# Centering the title\nax.set_title(\"Bronze Coverage in the US\", fontsize=20, pad=20)\n\n# Removing the axis for a cleaner look\nax.axis('off')\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Gold Map\nfig, ax = plt.subplots(figsize=(15, 10))  # Adjust the size here for a bigger map\n\nmerged_data.boundary.plot(ax=ax, color='black', linewidth=0.5)\nmerged_data.plot(column='Gold', ax=ax, legend=True, cmap='YlOrBr', legend_kwds={'label': \"Gold  Coverage\"})\n\n# Bounds to focus on the contiguous US\nax.set_xlim(-130, -60)\nax.set_ylim(20, 50)\n\n# Centering the title\nax.set_title(\"Gold  Coverage in the US\", fontsize=20, pad=20)\n\n# Removing the axis for a cleaner look\nax.axis('off')\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Silver Map\nfig, ax = plt.subplots(figsize=(15, 10))  # Adjust the size here for a bigger map\n\nmerged_data.boundary.plot(ax=ax, color='black', linewidth=0.5)\nmerged_data.plot(column='Silver', ax=ax, legend=True, cmap='Greens', legend_kwds={'label': \"Silver  Coverage\"})\n\n# Bounds to focus on the contiguous US\nax.set_xlim(-130, -60)\nax.set_ylim(20, 50)\n\n# Centering the title\nax.set_title(\"Silver  Coverage in the US\", fontsize=20, pad=20)\n\n# Removing the axis for a cleaner look\nax.axis('off')\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Rate by coverage level\n#Rate csv\ndf_rate=spark.read.csv(\"raw_data/Rate.csv\", header=True, inferSchema=True)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Subset of 2015 individual rate\ndf_rate2015 = df_rate.filter((df_rate.BusinessYear == 2015) & (df_rate.Age != \"Family Option\") & (df_rate.IndividualRate < 9000))\ndf_rate2015.show(5)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Subset the planAttributes and clean PlanId\ndf_benefit = df_planAttributes.select(df_planAttributes.columns[114], df_planAttributes.columns[103],df_planAttributes.columns[161], df_planAttributes.columns[165], df_planAttributes.columns[169])\ndf_benefit = df_benefit.withColumn(\"PlanId\", F.substring(df_benefit.PlanId, 1, 14))\ndf_rate2015 = df_rate2015.withColumn(\"PlanId\", df_rate2015.PlanId.cast(\"string\"))","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df_benefit.columns","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Clean MOOP columns\ndef clean_moop(column):\n    return regexp_replace(regexp_replace(F.col(column), \",\", \"\"), \"\\\\$\", \"\").cast(\"double\")\n\ndf_benefit = df_benefit.withColumn(\"TEHBInnTier1IndividualMOOP\", clean_moop(\"TEHBInnTier1IndividualMOOP\"))\ndf_benefit = df_benefit.withColumn(\"TEHBInnTier2IndividualMOOP\", clean_moop(\"TEHBInnTier2IndividualMOOP\"))\ndf_benefit = df_benefit.withColumn(\"TEHBOutOfNetIndividualMOOP\", clean_moop(\"TEHBOutOfNetIndividualMOOP\"))\n\n# Aggregate the benefit table\nbenefittouse = df_benefit.groupBy(\"PlanId\", \"MetalLevel\")\\\n.agg(\n    F.mean(\"TEHBInnTier1IndividualMOOP\").alias(\"innettier1moop\"),\n    F.mean(\"TEHBInnTier2IndividualMOOP\").alias(\"innettier2moop\"),\n    F.mean(\"TEHBOutOfNetIndividualMOOP\").alias(\"outnetmoop\")\n)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Join the benefit to the rates\ndf_planrates = df_rate2015.join(benefittouse, \"PlanId\", \"inner\")\n\n# Group by state and metal level, then aggregate\nbystatecoverage = df_planrates.groupBy(\"StateCode\", \"MetalLevel\").agg(\n    F.countDistinct(\"PlanId\").alias(\"PlanOffered\"),\n    F.mean(\"IndividualRate\").alias(\"MeanIndRate\"),\n    F.expr(\"percentile(IndividualRate, 0.5)\").alias(\"MedianIndRate\")  # median calculation\n).orderBy(F.desc(\"PlanOffered\"))\n\n# Exclude dental coverage\nmedicalonly = bystatecoverage.filter((bystatecoverage.MetalLevel != \"High\") & (bystatecoverage.MetalLevel != \"Low\"))","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"medicalonly = df_planrates\n\n# Aggregate data\nplan_counts = medicalonly.groupBy(\"StateCode\", \"MetalLevel\").count().toPandas()\n\nplt.figure(figsize=(10,6))\nsns.barplot(x='StateCode', y='count', hue='MetalLevel', data=plan_counts, errorbar=None)\nplt.title(\"Number of Medical Plans Offered By Coverage Levels\")\nplt.xlabel(\"Coverage Metal Level\")\nplt.ylabel(\"Number of Plans Offered\")\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Average Monthly Premium by Age","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\nbyagecoverage = df_planrates.groupBy(\"Age\", \"MetalLevel\") \\\n    .agg(\n        F.mean(\"IndividualRate\").alias(\"MeanIndRate\"),\n        F.expr('percentile_approx(IndividualRate, 0.5)').alias('MedianIndRate')\n    ) \\\n    .orderBy(F.desc(\"MeanIndRate\"))\n\nbyagecoverage_pd = byagecoverage.toPandas()\n\nplt.figure(figsize=(30,6))\nsns.scatterplot(x='Age', y='MeanIndRate', hue='MetalLevel', data=byagecoverage_pd, palette='deep')\nplt.title(\"Average Monthly Premium by Age\")\nplt.xlabel(\"Age\")\nplt.ylabel(\"Average Premium by Coverage\")\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Premium Distribution by Coverage Levels in 2015","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Due to limit memory I only presented the 20% of the data\nsample_fraction = 0.2\nplanrates_sampled = df_planrates.sample(False, sample_fraction)\n\n# Identify columns of type TimestampType\ntimestamp_cols = [f.name for f in planrates_sampled.schema.fields if isinstance(f.dataType, TimestampType)]\n\n# Convert to StringType\nfor col in timestamp_cols:\n    planrates_sampled = planrates_sampled.withColumn(col, planrates_sampled[col].cast(StringType()))\n\n# Convert the Spark dataframe to Pandas\nplanrates_pd = planrates_sampled.toPandas()\n\n# Convert the string date columns back to datetime64[ns] in Pandas\nfor col in timestamp_cols:\n    planrates_pd[col] = planrates_pd[col].astype('datetime64[ns]')\n\nplt.figure(figsize=(10,6))\nsns.boxplot(x='MetalLevel', y='IndividualRate', data=planrates_pd)\nsns.pointplot(x='MetalLevel', y='IndividualRate', data=planrates_pd, errorbar=None, estimator=np.mean, color='darkred')\nplt.title(\"Individual Rate Distribution of Coverage\")\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Determination the plans offered  across various US states","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df_benefit = spark.read.csv(\"raw_data/BenefitsCostSharing.csv\", header=True, inferSchema=True)\nnum_fields = len(df_benefit.columns)\nprint(f\"Number of fields: {num_fields}\")\n\n# Remove empty or null rows\ndf_no_nulls = df_benefit.dropna()\nfor column in df_no_nulls.columns:\n    df_no_nulls = df_no_nulls.filter(F.col(column) != \"\")  # Use F.col here\n\nnum_rows = df_benefit.count()\nprint(f\"The CSV file has {num_rows} rows.\")","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Subset for planRateBenefit\ndf_planRateBenefit = df_rate\\\n.filter((df_rate[\"IndividualRate\"] < 9999) & (df_rate[\"Age\"] != \"Family Option\"))\\\n.select(\"BusinessYear\", \"StateCode\", \"IssuerId\", \"PlanId\", \"Age\", \"IndividualRate\", \"IndividualTobaccoRate\")\n\ndf_planRateBenefit.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df_StateCarrier = df_planRateBenefit.groupBy(\"StateCode\", \"IssuerId\", \"PlanId\")\\\n.agg(count(\"StateCode\").alias(\"count\"))\n\ndf_StateCarrier.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Graph\ndataBenefitsName = df_benefit.select(\"BusinessYear\", \"IssuerId\", \"StandardComponentId\", \"StateCode\", \"BenefitName\")\ndataBenefitsName.show(5)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Subset for VariousRatesPerState\nVariousRatesPerState = df_rate\\\n.filter(\n    (df_rate[\"Couple\"].isNotNull()) | \n    (df_rate[\"PrimarySubscriberAndOneDependent\"].isNotNull()) |\n    (df_rate[\"PrimarySubscriberAndTwoDependents\"].isNotNull()) |\n    (df_rate[\"PrimarySubscriberAndThreeOrMoreDependents\"].isNotNull()) |\n    (df_rate[\"CoupleAndOneDependent\"].isNotNull()) |\n    (df_rate[\"CoupleAndTwoDependents\"].isNotNull()) |\n    (df_rate[\"CoupleAndThreeOrMoreDependents\"].isNotNull()))\\\n.select(\"StateCode\", \"IndividualRate\", \"Couple\", \"PrimarySubscriberAndOneDependent\", \n         \"PrimarySubscriberAndTwoDependents\", \"PrimarySubscriberAndThreeOrMoreDependents\", \n         \"CoupleAndOneDependent\", \"CoupleAndTwoDependents\", \"CoupleAndThreeOrMoreDependents\")\nVariousRatesPerState.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"TotalRatePerState = VariousRatesPerState.groupBy(\n    \"StateCode\", \"IndividualRate\", \"Couple\", \"PrimarySubscriberAndOneDependent\", \n    \"PrimarySubscriberAndTwoDependents\", \"PrimarySubscriberAndThreeOrMoreDependents\", \n    \"CoupleAndOneDependent\", \"CoupleAndTwoDependents\", \"CoupleAndThreeOrMoreDependents\")\\\n.agg(count(\"StateCode\").alias(\"count\"))\n\n\nTotalRatePerState.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Melt operation \nmelted_columns = [\"IndividualRate\", \"Couple\", \"PrimarySubscriberAndOneDependent\", \n                  \"PrimarySubscriberAndTwoDependents\", \"PrimarySubscriberAndThreeOrMoreDependents\", \n                  \"CoupleAndOneDependent\", \"CoupleAndTwoDependents\", \"CoupleAndThreeOrMoreDependents\"]\ndf_melt = TotalRatePerState.select(\"StateCode\", F.expr('stack(' + str(len(melted_columns)) + ', ' + ', '.join([\"'\" + x + \"', \" + x for x in melted_columns]) + ') as (RateType, RateValue)'))\ndf_melt.show(5)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# PlanRatevsAge subset\nPlanRatevsAge = df_rate.filter(\n    (df_rate[\"IndividualRate\"] < 9999) & \n    (df_rate[\"StateCode\"].isNotNull()) & \n    (df_rate[\"Age\"].isNotNull()) & \n    (df_rate[\"IndividualRate\"].isNotNull())\n).select(\"StateCode\", \"Age\", \"IndividualRate\")\n\n# Filter data for different age groups and drop duplicates based on 'IndividualRate'\nTotRatevsAge1 = PlanRatevsAge.filter(PlanRatevsAge[\"Age\"] <= 22).dropDuplicates([\"IndividualRate\"])\nTotRatevsAge2 = PlanRatevsAge.filter((PlanRatevsAge[\"Age\"] >= 23) & (PlanRatevsAge[\"Age\"] <= 40)).dropDuplicates([\"IndividualRate\"])\nTotRatevsAge3 = PlanRatevsAge.filter((PlanRatevsAge[\"Age\"] >= 41) & (PlanRatevsAge[\"Age\"] <= 59)).dropDuplicates([\"IndividualRate\"])\nTotRatevsAge4 = PlanRatevsAge.filter(PlanRatevsAge[\"Age\"] == \"Family Option\").dropDuplicates([\"IndividualRate\"])\nTotRatevsAge5 = PlanRatevsAge.filter((PlanRatevsAge[\"Age\"] >= 60) & (PlanRatevsAge[\"Age\"] <= 64)).dropDuplicates([\"IndividualRate\"])\nTotRatevsAge6 = PlanRatevsAge.filter(PlanRatevsAge[\"Age\"] == \"65 and over\").dropDuplicates([\"IndividualRate\"])","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Convert a PySpark DataFrame to Pandas for plotting\npdf = df_melt.toPandas()\n\n# Sort the DataFrame by 'StateCode'\npdf = pdf.sort_values(by='StateCode')\n\n# Split the state codes into two halves\nunique_states = sorted(pdf['StateCode'].unique())\nhalf_length = len(unique_states) // 2\n\nfirst_half_states = unique_states[:half_length]\nsecond_half_states = unique_states[half_length:]\n\n# Plot the first half\nplt.figure(figsize=(15, 8))\nsns.barplot(data=pdf[pdf['StateCode'].isin(first_half_states)], x='StateCode', y='RateValue', hue='RateType', order=first_half_states)\nplt.title(\"Plot of Health insurance Plan rates\")\nplt.ylabel(\"Plan rates\")\nplt.xlabel(\"State Code\")\nplt.xticks(rotation=90)\nplt.legend(title='Rate Plans')\nplt.tight_layout()\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Plot the second half\nplt.figure(figsize=(15, 8))\nsns.barplot(data=pdf[pdf['StateCode'].isin(second_half_states)], x='StateCode', y='RateValue', hue='RateType', order=second_half_states)\nplt.title(\"Plot of Health insurance Plan rates\")\nplt.ylabel(\"Plan rates\")\nplt.xlabel(\"State Code\")\nplt.xticks(rotation=90)\nplt.legend(title='Rate Plans')\nplt.tight_layout()\nplt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Graphs per age\npdf1 = TotRatevsAge1.toPandas()\npdf2 = TotRatevsAge2.toPandas()\npdf3 = TotRatevsAge3.toPandas()\npdf4 = TotRatevsAge4.toPandas()\npdf5 = TotRatevsAge5.toPandas()\npdf6 = TotRatevsAge6.toPandas()\n #Function for creating all the plots be group age\ndef plot_rate_vs_age(data, title):\n    plt.figure(figsize=(50, 10))\n    sns.barplot(data=data, x='StateCode', y='IndividualRate', hue='Age')\n    plt.ylabel(\"Individual Rates\")\n    plt.xlabel(\"State Code\")\n    plt.title(title)\n    plt.xticks(rotation=90)\n    plt.legend(title='Age')\n    plt.tight_layout()\n    plt.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Age 0-22\nplot_rate_vs_age(pdf1, \"Plan Rate vs Age across various US States\")","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Age 23-40\nplot_rate_vs_age(pdf2, \"Plan Rate vs Age across various US States\")","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Age 41 to 59 \nplot_rate_vs_age(pdf3, \"Plan Rate vs Age across various US States\")","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Age 60 to 64\nplot_rate_vs_age(pdf5, \"Plan Rate vs Age across various US States\")","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Age 65 and over\nplot_rate_vs_age(pdf6, \"Plan Rate vs Age across various US States\")","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Family edition\nplot_rate_vs_age(pdf4, \"Plan Rate vs Age across various US States\")","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#DentalCare System","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Prepare Plan Attributes with Dental data","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Keep only the columns with dental data\nselected_columns = (\n    df_planAttributes.columns[0:4] +\n    [df_planAttributes.columns[5], df_planAttributes.columns[6]] +\n    df_planAttributes.columns[9:15] +\n    df_planAttributes.columns[16:21] +\n    df_planAttributes.columns[24:28] +\n    df_planAttributes.columns[29:50] +\n    df_planAttributes.columns[54:125]\n)\ndf_cleanPlan=df_planAttributes.select(selected_columns)\n\n#Remove\nrm_cols=[\"OutOfCountryCoverageDescription\",\"OutOfServiceAreaCoverageDescription\", \"DiseaseManagementProgramsOffered\",\"PlanEffictiveDate\", \"PlanExpirationDate\", \"OutOfCountryCoverage\",\n        \"MedicalDrugMaximumOutofPocketIntegrated\", \"FirstTierUtilization\",\"SecondTierUtilization\",\"MEHBInnTier1IndividualMOOP\",\"MEHBInnTier1FamilyMOOP\",\"MEHBInnTier2IndividualMOOP\",\n        \"MEHBInnTier2FamilyMOOP\",\"MEHBOutOfNetIndividualMOOP\",\"MEHBOutOfNetFamilyMOOP\",\"MEHBCombInnOonIndividualMOOP\",\n         \"MEHBCombInnOonFamilyMOOP\",\"DEHBInnTier1IndividualMOOP\",\"DEHBInnTier1FamilyMOOP\",\"DEHBInnTier2IndividualMOOP\",\"DEHBInnTier2FamilyMOOP\",\"DEHBOutOfNetIndividualMOOP\",\n         \"DEHBOutOfNetFamilyMOOP\",\"DEHBCombInnOonIndividualMOOP\",\"DEHBCombInnOonFamilyMOOP\",\"TEHBInnTier1IndividualMOOP\",                              \n        \"TEHBInnTier1FamilyMOOP\",\"TEHBInnTier2IndividualMOOP\",\"TEHBInnTier2FamilyMOOP\",\"TEHBOutOfNetIndividualMOOP\",                              \n        \"TEHBOutOfNetFamilyMOOP\",\"TEHBCombInnOonIndividualMOOP\",\"TEHBCombInnOonFamilyMOOP\",\"MEHBDedInnTier1Individual\",                               \n        \"MEHBDedInnTier1Family\",\"MEHBDedInnTier1Coinsurance\",\"MEHBDedInnTier2Individual\",\"MEHBDedInnTier2Family\",\"MEHBDedInnTier2Coinsurance\",\n        \"MEHBDedOutOfNetIndividual\",\"MEHBDedOutOfNetFamily\",\"MEHBDedCombInnOonIndividual\",\"MEHBDedCombInnOonFamily\",\"DEHBDedInnTier1Individual\",\n         \"DEHBDedInnTier1Family\",\"DEHBDedInnTier1Coinsurance\", \"DEHBDedInnTier2Individual\",\"DEHBDedInnTier2Family\",                                   \n        \"DEHBDedInnTier2Coinsurance\",\"DEHBDedOutOfNetIndividual\",\"DEHBDedOutOfNetFamily\",\"DEHBDedCombInnOonIndividual\",                             \n        \"DEHBDedCombInnOonFamily\",\"TEHBDedInnTier1Individual\",\"TEHBDedInnTier1Family\",\"TEHBDedInnTier1Coinsurance\",                              \n        \"TEHBDedInnTier2Individual\",\"TEHBDedInnTier2Family\",\"TEHBDedInnTier2Coinsurance\",\"TEHBDedOutOfNetIndividual\",                              \n        \"TEHBDedOutOfNetFamily\",\"TEHBDedCombInnOonIndividual\",\"TEHBDedCombInnOonFamily\",\"SBCHavingaBabyDeductible\",                                \n        \"SBCHavingaBabyCopayment\",\"SBCHavingaBabyCoinsurance\",\"SBCHavingaBabyLimit\",\"SBCHavingDiabetesDeductible\",                            \n        \"SBCHavingDiabetesCopayment\",\"SBCHavingDiabetesCoinsurance\",\"SBCHavingDiabetesLimit\",\n        \"FormularyId\", \"MedicalDrugDeductiblesIntegrated\", \"InpatientCopaymentMaximumDays\",\n        \"SpecialtyDrugMaximumCoinsurance\", \"IsNoticeRequiredForPregnancy\"]\ndf_cleanPlan=df_planAttributes.drop(*rm_cols)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df_cleanPlan=df_cleanPlan.withColumn(\"plan_id_short\", \n                                     F.expr(\"substring(PlanId, 1, length(PlanId) - 3)\"))\n# Filter to get dental_only_list and not_dental_only_list\ndental_only_list = df_cleanPlan.filter(df_cleanPlan.DentalOnlyPlan == \"Yes\").select(\"PlanId\",df_cleanPlan.columns[2], df_cleanPlan.columns[3], df_cleanPlan.columns[4], df_cleanPlan.columns[11], df_cleanPlan.columns[33])\n\n# Count unique PlanId\nprint(dental_only_list.select(\"PlanId\").distinct().count())","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Prepare 'Benefits of Cost Sharing' with Dental data","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Load BenefitsCostSharing\ndf_costShare = spark.read.csv(\"raw_data/BenefitsCostSharing.csv\", header=True, inferSchema=True)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"column_positions = {column: index for index, column in enumerate(df_costShare.columns)}\nprint(column_positions)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"all_columns = df_costShare.columns\nselect_bnf_cols =df_costShare.columns[0:8] + df_costShare.columns[9:31]\ndf_cleanbnf=df_costShare.select(select_bnf_cols)\ndf_cleanbnf.printSchema()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Linked benefits to dentalcare","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#General examination\ndf_cleanbnf = df_cleanbnf.withColumn(\n    \"routineexam\",\n    F.when(\n        (F.col(\"BenefitName\").rlike(\"Prophylaxis|Oral Exam|Cleaning|Dental Check-Up for Children\")) & \n        (F.col(\"IsCovered\") == \"Covered\"),\n        1\n    ).otherwise(None)\n)\n\n# Summary table\ngeneral_exams = df_cleanbnf.filter(F.col(\"routineexam\") > 0).groupBy(\"BenefitName\").count()\ngeneral_exams = general_exams.withColumnRenamed(\"count\", \"Covering Plans\")\ngeneral_exams.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Basic Denatalcare\ndf_cleanbnf = df_cleanbnf.withColumn(\n    \"basicdental\",\n    F.when(\n        (F.col(\"BenefitName\").rlike(\"Basic Dental Care|Routine Dental Services\")) & \n        (F.col(\"IsCovered\") == \"Covered\"),\n        1\n    ).otherwise(\n        F.when(\n            (F.col(\"BenefitName\").rlike(\"Non-Routine\")) & \n            (F.col(\"IsCovered\") == \"Covered\"),\n            0\n        ).otherwise(None)\n    )\n)\n\n# Summary table\nbasic_dental = df_cleanbnf.filter(F.col(\"basicdental\") > 0).groupBy(\"BenefitName\").count()\nbasic_dental = basic_dental.withColumnRenamed(\"count\", \"Covering Plans\")\nbasic_dental.show()\n","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Basic Dental Adult\nif 'basicdental_adult' not in df_cleanbnf.columns:\n    df_cleanbnf = df_cleanbnf.withColumn('basicdental_adult', F.lit(0))\n\n# Your existing code\ndf_cleanbnf = df_cleanbnf.withColumn(\n    \"basicdental_adult\", \n    F.when(\n        (F.col(\"BenefitName\").like(\"%Basic Dental Care - Adult%\")) & (F.col(\"IsCovered\") == \"Covered\"), \n        1\n    ).otherwise(F.col(\"basicdental_adult\"))\n)\n# Summary table\nbasicdental_adult = df_cleanbnf.filter(F.col(\"basicdental_adult\") > 0).groupBy(\"BenefitName\").count()\nbasicdental_adult = basic_dental.withColumnRenamed(\"count\", \"Covering Plans\")\nbasicdental_adult.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Major Dentalcare\ndf_cleanbnf = df_cleanbnf.withColumn(\n    \"majordental\",\n    F.when(\n        (F.col(\"BenefitName\").rlike(\"Major Dental Care|Non-Routine\")) & \n        (F.col(\"IsCovered\") == \"Covered\"),\n        1\n    ).otherwise(None)\n)\n\n# Summary table\nmajor_dental = df_cleanbnf.filter(F.col(\"majordental\") > 0).groupBy(\"BenefitName\").count()\nmajor_dental = major_dental.withColumnRenamed(\"count\", \"Covering Plans\")\nmajor_dental.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#X-Rays\ndf_cleanbnf = df_cleanbnf.withColumn(\n    \"xrays\",\n    F.when(\n        (F.col(\"BenefitName\").rlike(\"X-ray|X-Ray|X ray|X Ray\")) & \n        (F.col(\"IsCovered\") == \"Covered\"),\n        1\n    ).otherwise(\n        F.when(\n            F.col(\"BenefitName\") == \"X-rays and Diagnostic Imaging\",\n            0\n        ).otherwise(None)\n    )\n)\n\n# Summary table\nxrays = df_cleanbnf.filter(F.col(\"xrays\") > 0).groupBy(\"BenefitName\").count()\nxrays = xrays.withColumnRenamed(\"count\", \"Covering Plans\")\nxrays.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Fluoride Treatment\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'fluoride',\n    F.when(\n        (F.col('BenefitName').rlike('Flouride|Fluoride|fluoride')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n\n# Summary table\nfluoride = df_cleanbnf.filter(F.col(\"fluoride\") > 0).groupBy(\"BenefitName\").count()\nfluoride = fluoride.withColumnRenamed(\"count\", \"Covering Plans\")\nfluoride.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Extranctions\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'extract',\n    F.when(\n        (F.col('BenefitName').rlike('Extraction|extraction|removal|Removal')) & (F.col('IsCovered') == 'Covered') & (F.col('BenefitName') != 'Breast Implant Removal'),\n        1\n    ).otherwise(None)\n)\n# Summary table\nextract = df_cleanbnf.filter(F.col(\"extract\") > 0).groupBy(\"BenefitName\").count()\nextract = extract.withColumnRenamed(\"count\", \"Covering Plans\")\nextract.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Root Canals\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'rootcanal',\n    F.when(\n        (F.col('BenefitName').rlike('Root Canal|root canal|Root canal|Endodonti')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\nrootcanal = df_cleanbnf.filter(F.col('rootcanal') > 0).groupBy('BenefitName').count()\nrootcanal = rootcanal.withColumnRenamed('count', 'Covering Plans')\nrootcanal.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Sealants\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'sealant',\n    F.when(\n        (F.col('BenefitName').rlike('Sealant|sealant')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\nsealant = df_cleanbnf.filter(F.col('sealant') > 0).groupBy('BenefitName').count()\nsealant = sealant.withColumnRenamed('count', 'Covering Plans')\nsealant.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Fillings\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'fillings',\n    F.when(\n        (F.col('BenefitName').rlike('restorative|Restorative|Amalgam|Filling')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\nfillings = df_cleanbnf.filter(F.col('fillings') > 0).groupBy('BenefitName').count()\nfillings = fillings.withColumnRenamed('count', 'Covering Plans')\nfillings.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Periodontics\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'periodontics',\n    F.when(\n        (F.col('BenefitName').rlike('Periodont|periodont|Scaling|Planing')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\nperiodontics = df_cleanbnf.filter(F.col('periodontics') > 0).groupBy('BenefitName').count()\nperiodontics = periodontics.withColumnRenamed('count', 'Covering Plans')\nperiodontics.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Orthodontia\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'orthodontia',\n    F.when(\n        (F.col('BenefitName').rlike('Orthognatic|orthodont|Orthodont|Occlusal|Ocalusal|occlusal')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\northodontia = df_cleanbnf.filter(F.col('orthodontia') > 0).groupBy('BenefitName').count()\northodontia = orthodontia.withColumnRenamed('count', 'Covering Plans')\northodontia.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Dentures or Partials\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'denture',\n    F.when(\n        (F.col('BenefitName').rlike('Denture|denture|Bridge|Dentiures')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\ndenture = df_cleanbnf.filter(F.col('denture') > 0).groupBy('BenefitName').count()\ndenture = denture.withColumnRenamed('count', 'Covering Plans')\ndenture.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Specialist Gum Procedures\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'gums',\n    F.when(\n        (F.col('BenefitName').rlike('Gingivec')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\ngums = df_cleanbnf.filter(F.col('gums') > 0).groupBy('BenefitName').count()\ngums = gums.withColumnRenamed('count', 'Covering Plans')\ngums.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Dental Anesthesia\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'anesthesia',\n    F.when(\n        (F.col('BenefitName').rlike('Dental Anesthesia')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\nanesthesia = df_cleanbnf.filter(F.col('anesthesia') > 0).groupBy('BenefitName').count()\nanesthesia = anesthesia.withColumnRenamed('count', 'Covering Plans')\nanesthesia.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Crowns\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'crown',\n    F.when(\n        (F.col('BenefitName').rlike('Crown')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\ncrown = df_cleanbnf.filter(F.col('crown') > 0).groupBy('BenefitName').count()\ncrown = crown.withColumnRenamed('count', 'Covering Plans')\ncrown.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Cosmetic Dentistry\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'cosmetic',\n    F.when(\n        (F.col('BenefitName').rlike('Cosmetic Ortho')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\ncosmetic = df_cleanbnf.filter(F.col('cosmetic') > 0).groupBy('BenefitName').count()\ncosmetic = cosmetic.withColumnRenamed('count', 'Covering Plans')\ncosmetic.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Cosmetic Dentistry\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'cosmetic',\n    F.when(\n        (F.col('BenefitName').rlike('Cosmetic Ortho')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\ncosmetic = df_cleanbnf.filter(F.col('cosmetic') > 0).groupBy('BenefitName').count()\ncosmetic = cosmetic.withColumnRenamed('count', 'Covering Plans')\ncosmetic.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Dental Surgery\ndf_cleanbnf = df_cleanbnf.withColumn(\n    'surgery',\n    F.when(\n        (F.col('BenefitName').rlike('Oral Surgery|Orthognatic Treatment/Surgery|Osseous Surgery')) & (F.col('IsCovered') == 'Covered'),\n        1\n    ).otherwise(None)\n)\n# Summary table\nsurgery = df_cleanbnf.filter(F.col('surgery') > 0).groupBy('BenefitName').count()\nsurgery = surgery.withColumnRenamed('count', 'Covering Plans')\nsurgery.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Dental benefits summary","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"ben_summary = df_cleanbnf.groupBy('PlanId', 'IssuerId').agg(\n    F.sum('xrays').alias('xrays'),\n    F.sum('fluoride').alias('fluoride'),\n    F.sum('majordental').alias('majordental'),\n    F.sum('extract').alias('extract'),\n    F.sum('rootcanal').alias('rootcanal'),\n    F.sum('sealant').alias('sealant'),\n    F.sum('routineexam').alias('routineexam'),\n    F.sum('basicdental').alias('basicdental'),\n    F.sum('basicdental_adult').alias('basicdental_adult'),\n    F.sum('fillings').alias('fillings'),\n    F.sum('periodontics').alias('periodontics'),\n    F.sum('orthodontia').alias('orthodontia'),\n    F.sum('denture').alias('denture'),\n    F.sum('gums').alias('gums'),\n    F.sum('anesthesia').alias('anesthesia'),\n    F.sum('crown').alias('crown'),\n    F.sum('cosmetic').alias('cosmetic'),\n    F.sum('surgery').alias('surgery')\n)\nben_summary.show(5)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#The amount a patient pays out of pocket for a visit to the dentist.\n#If it's a flat dollar amount, that's a copay. \n#If it's a percentage of the charges, it's coinsurance. Together, this is \"cost sharing\" in insurance parlance.","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"ben_costs=df_cleanbnf.join(ben_summary.select(\"PlanId\"),on=\"PlanId\", how=\"inner\").distinct()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Clean and convert Copay strings to numeric\nfor column in [\"CopayInnTier1\", \"CopayInnTier2\", \"CopayOutofNet\"]:\n    # Remove unwanted substrings\n    for substring in [\"Copay before deductible\", \"Copay after deductible\", \"Copay per Stay\", \"Copay per Day\"]:\n        ben_costs = ben_costs.withColumn(column, F.expr(f\"TRIM(REPLACE({column}, '{substring}', ''))\"))\n    # Remove commas and dollar signs\n    ben_costs = ben_costs.withColumn(column, F.expr(f\"TRIM(REPLACE(REPLACE({column}, ',', ''), '$', ''))\"))\n    # Convert to numeric\n    ben_costs = ben_costs.withColumn(column + \"_num\", ben_costs[column].cast(DoubleType()))\n    # Handle special values\n    ben_costs = ben_costs.withColumn(column + \"_num\", F.when(ben_costs[column].contains(\"No Charge\") | (ben_costs[column] == \"\") | (ben_costs[column] == \"0%\"), 0).otherwise(ben_costs[column + \"_num\"]))","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Generate summary tables\nben_costs.select(\"CopayInnTier1_num\", \"CopayInnTier2_num\", \"CopayOutofNet_num\").summary().show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Histogram for CoinsINNTier1_num\nben_costs1 = ben_costs.filter((ben_costs.CoinsInnTier1_num < 1000) & (ben_costs.CoinsInnTier1_num > 0))\nben_costs1_pd = ben_costs1.select(\"CoinsInnTier1_num\").toPandas()\nben_costs1_pd.plot(kind='hist', bins=range(0, 1000, 5), edgecolor='black')","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Histogram for CoinsINNTier2_num\nben_costs1 = ben_costs.filter((ben_costs.CoinsInnTier2_num < 1000) & (ben_costs.CoinsInnTier2_num > 0))\nben_costs1_pd = ben_costs1.select(\"CoinsInnTier2_num\").toPandas()\nben_costs1_pd.plot(kind='hist', bins=range(0, 1000, 5), edgecolor='black')","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Filter and plot for CoinsOutofNet_num\nben_costs1 = ben_costs.filter((ben_costs.CoinsOutofNet_num< 1000) & (ben_costs.CoinsOutofNet_num > 0))\nben_costs1_pd = ben_costs1.select(\"CoinsOutofNet_num\").toPandas()\nben_costs1_pd.plot(kind='hist', bins=range(0, 1000, 5), edgecolor='black')","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Filter and plot for CopayInnTier1_num\nben_costs1 = ben_costs.filter((ben_costs.CopayInnTier1_num< 1000) & (ben_costs.CopayInnTier1_num > 0))\nben_costs1_pd = ben_costs1.select(\"CopayInnTier1_num\").toPandas()\nben_costs1_pd.plot(kind='hist', bins=range(0, 1000, 5), edgecolor='black')","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Filter and plot for CopayInnTier2_num\nben_costs1 = ben_costs.filter((ben_costs.CopayInnTier2_num< 1000) & (ben_costs.CopayInnTier2_num > 0))\nben_costs1_pd = ben_costs1.select(\"CopayInnTier2_num\").toPandas()\nben_costs1_pd.plot(kind='hist', bins=range(0, 1000, 5), edgecolor='black')","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Filter and plot for CopayOutofNet_num\nben_costs1 = ben_costs.filter((ben_costs.CopayOutofNet_num< 1000) & (ben_costs.CopayOutofNet_num> 0))\nben_costs1_pd = ben_costs1.select(\"CopayOutofNet_num\").toPandas()\nben_costs1_pd.plot(kind='hist', bins=range(0, 1000, 5), edgecolor='black')","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Benefit of cost sharing for the dental categories\n#Basic Dental\nben_costs = ben_costs.withColumn(\n    \"basicdental\", \n    F.when(\n        (F.col(\"BenefitName\").rlike(\"Basic Dental Care|Routine Dental Services\")) & (F.col(\"IsCovered\") == \"Covered\"),\n        1\n    ).otherwise(0)\n)\n\n#Basicdental_adult\nben_costs = ben_costs.withColumn(\n    \"basicdental_adult\", \n    F.when(\n        (F.col(\"BenefitName\").rlike(\"Basic Dental Care|Adult\")) & (F.col(\"IsCovered\") == \"Covered\"),\n        1\n    ).otherwise(0)\n)\n#Majordental\nben_costs = ben_costs.withColumn(\n    \"majordental\", \n    F.when(\n        (F.col(\"BenefitName\").rlike(\"Major Dental Care\")) & (F.col(\"IsCovered\") == \"Covered\"),\n        1\n    ).otherwise(0)\n)\n\n#Fillings\nben_costs = ben_costs.withColumn(\n    \"fillings\", \n    F.when(\n        (F.col(\"BenefitName\").rlike(\"restorative|Restorative|Amalgam|Filling\")) & (F.col(\"IsCovered\") == \"Covered\"),\n        1\n    ).otherwise(0)\n)\n\n#Extract\nben_costs = ben_costs.withColumn(\n    \"extract\", \n    F.when(F.col(\"BenefitName\") == \"Breast Implant Removal\", 0).otherwise(F.col(\"extract\"))\n)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Defination the common condition expressions to reuse\ncondition_basicdental = F.col(\"basicdental\") == 1\ncondition_basicdental_adult = F.col(\"basicdental_adult\") == 1\ncondition_majordental = F.col(\"majordental\") == 1\ncondition_fillings = F.col(\"fillings\") == 1\ncondition_extract = F.col(\"extract\") == 1\n\nagg_exprs = []","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Helper function to add aggregation expressions for Copay\ndef add_agg_for_copay(condition, alias_prefix):\n    agg_exprs.extend([\n        F.sum(F.when(condition, F.col(\"CopayInnTier1_num\")).otherwise(0)).alias(f\"{alias_prefix}_copay_inn_1\"),\n        F.sum(F.when(condition, F.col(\"CopayInnTier2_num\")).otherwise(0)).alias(f\"{alias_prefix}_copay_inn_2\"),\n        F.sum(F.when(condition, F.col(\"CopayOutofNet_num\")).otherwise(0)).alias(f\"{alias_prefix}_copay_oon\"),\n    ])\n\n# Helper function to add aggregation expressions for Coinsurance\ndef add_agg_for_coinsurance(condition, alias_prefix):\n    agg_exprs.extend([\n        F.sum(F.when(condition, F.col(\"CoinsINNTier1_num\")).otherwise(0)).alias(f\"{alias_prefix}_coin_inn_1\"),\n        F.sum(F.when(condition, F.col(\"CoinsINNTier2_num\")).otherwise(0)).alias(f\"{alias_prefix}_coin_inn_2\"),\n        F.sum(F.when(condition, F.col(\"CoinsOutofNet_num\")).otherwise(0)).alias(f\"{alias_prefix}_coin_oon\"),\n    ])","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Using helper functions to add aggregation expressions\nadd_agg_for_copay(condition_basicdental, \"basic\")\nadd_agg_for_copay(condition_basicdental_adult, \"basic_a\")\nadd_agg_for_copay(condition_majordental, \"majordental\")\nadd_agg_for_copay(condition_fillings, \"fillings\")\nadd_agg_for_copay(condition_extract, \"extract\")\n\nadd_agg_for_coinsurance(condition_basicdental, \"basic\")\nadd_agg_for_coinsurance(condition_basicdental_adult, \"basic_a\")\nadd_agg_for_coinsurance(condition_majordental, \"majordental\")\nadd_agg_for_coinsurance(condition_fillings, \"fillings\")\nadd_agg_for_coinsurance(condition_extract, \"extract\")\n\n# Grouping and Aggregation\nben_cost_summary = ben_costs.groupBy(\"PlanId\", \"IssuerId\").agg(*agg_exprs)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Performing the first merge (join)\ndental_data = df_planAttributes.join(ben_summary, on=[\"PlanId\", \"IssuerId\"], how=\"inner\")\n\n# Performing the second merge (join)\ndental_data = dental_data.join(ben_cost_summary, on=[\"PlanId\", \"IssuerId\"], how=\"inner\")\n\n\n#dental_data.summary().show()\nprint(dental_data.columns)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Create Dental Data file","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Create a new column 'plan_id_short' using substring function\ndf_planAttributes_filtered = df_planAttributes.withColumn(\"plan_id_short\", F.substring(F.col(\"PlanId\"), 1, -4))\n\n\n# Define the join keys\njoin_keys = [\"PlanId\", \"IssuerId\"]\n\n# Perform the merge using the join keys\ndental_data = df_planAttributes_filtered.join(ben_summary, on=join_keys, how=\"inner\")\n\n# Show the merged data\noutput_path = \"output_files/dental_data.csv\"\ndental_data.write.csv(output_path, header=True, mode=\"overwrite\")","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Rates are the monthly payments regardless of whether you go see your dentist or not. ","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"rate=spark.read.csv(\"raw_data/Rate.csv\", header=True, inferSchema=True)\nrate.printSchema()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pyspark.sql.window import Window\n# Filter rate: remove rates that are either above 1000 or below 0\nrate = rate.filter((rate.IndividualRate < 1000) & (rate.IndividualRate > 0))\n\n# Load dental_data into a Spark DataFrame, adjust path as needed\ndental_data = spark.read.csv(\"output_files/dental_data.csv\", header=True, inferSchema=True)\n\n# Assuming PlanId is the common column between the two DataFrames\nrate = rate.join(dental_data.select(\"plan_id_short\", \"DentalOnlyPlan\"), rate.PlanId == dental_data.plan_id_short, how='left')\n\n# Create age groups and calculate mean IndividualRate for each group\nage_groups = {\n    \"0-20\":\"mean_rate_kids\",\n    \"21-29\":\"mean_rate_20s\",\n    \"30-39\":\"mean_rate_30s\",\n    \"40-49\":\"mean_rate_40s\",\n    \"50-64\":\"mean_rate_5064\",\n    \"65 and over\":\"mean_rate_65up\",\n    \"Family Option\":\"mean_rate_family\"\n}","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"for age_group, output_col in age_groups.items():\n    lower_age, upper_age = age_group.split(\"-\") if \"-\" in age_group else (age_group, age_group)\n    age_conditions = (F.col(\"Age\") >= lower_age) & (F.col(\"Age\") <= upper_age)\n\n    windowSpec = Window.partitionBy(\"PlanId\").orderBy(\"Age\").rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)\n    mean_rate_do = F.mean(F.when(age_conditions & (F.col(\"DentalOnlyPlan\") == \"Yes\"), F.col(\"IndividualRate\"))).over(windowSpec)\n    rate = rate.withColumn(output_col + \"_do\", mean_rate_do)\n\n\n    # For DentalOnlyPlan == \"No\"\n    mean_rate_dm = F.mean(F.when(age_conditions & (F.col(\"DentalOnlyPlan\") == \"No\"), F.col(\"IndividualRate\"))).over(windowSpec)\n    rate = rate.withColumn(output_col + \"_dm\", mean_rate_dm)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Calculating mean IndividualTobaccoRate for DentalOnlyPlan == \"Yes\"\nwindowSpec_do = Window.partitionBy(\"PlanId\").orderBy(\"DentalOnlyPlan\").rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)\nmean_rate_tobacco_do = F.mean(F.when(F.col(\"DentalOnlyPlan\") == \"Yes\", F.col(\"IndividualTobaccoRate\"))).over(windowSpec_do).alias(\"mean_rate_tobacco_do\")\nrate = rate.withColumn(\"mean_rate_tobacco_do\", mean_rate_tobacco_do)\n\n# Calculating mean IndividualTobaccoRate for DentalOnlyPlan == \"No\"\nmean_rate_tobacco_dm = F.mean(F.when(F.col(\"DentalOnlyPlan\") == \"No\", F.col(\"IndividualTobaccoRate\"))).over(windowSpec_do).alias(\"mean_rate_tobacco_dm\")\nrate = rate.withColumn(\"mean_rate_tobacco_dm\", mean_rate_tobacco_dm)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"rate_summary = rate.groupBy(\"PlanId\").agg(\n    mean(\"mean_rate_kids_do\").alias(\"mean_rate_kids_do\"),\n    mean(\"mean_rate_20s_do\").alias(\"mean_rate_20s_do\"),\n    mean(\"mean_rate_30s_do\").alias(\"mean_rate_30s_do\"),\n    mean(\"mean_rate_40s_do\").alias(\"mean_rate_40s_do\"),\n    mean(\"mean_rate_5064_do\").alias(\"mean_rate_5064_do\"),\n    mean(\"mean_rate_65up_do\").alias(\"mean_rate_65up_do\"),\n    mean(\"mean_rate_family_do\").alias(\"mean_rate_family_do\"),\n    mean(\"mean_rate_tobacco_do\").alias(\"mean_rate_tobacco_do\"),\n    mean(\"mean_rate_kids_dm\").alias(\"mean_rate_kids_dm\"),\n    mean(\"mean_rate_20s_dm\").alias(\"mean_rate_20s_dm\"),\n    mean(\"mean_rate_30s_dm\").alias(\"mean_rate_30s_dm\"),\n    mean(\"mean_rate_40s_dm\").alias(\"mean_rate_40s_dm\"),\n    mean(\"mean_rate_5064_dm\").alias(\"mean_rate_5064_dm\"),\n    mean(\"mean_rate_65up_dm\").alias(\"mean_rate_65up_dm\"),\n    mean(\"mean_rate_tobacco_dm\").alias(\"mean_rate_tobacco_dm\")\n)\n\nrate_summary.printSchema()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Read Network csv\nnetwork = spark.read.csv(\"raw_data/Network.csv\", header=True, inferSchema=True)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Merge dental_data with network\ndental_data = dental_data.join(network, on=[\"BusinessYear\", \"StateCode\", \"IssuerId\", \"SourceName\", \"NetworkId\"], how=\"left\")\n\n# Create a new column plan_id_short by extracting the substring\ndental_data = dental_data.withColumn(\"plan_id_short\", dental_data[\"PlanId\"].substr(1, 4))\n\n# Merge dental_data with rate_summary\ndental_data = dental_data.join(rate_summary, dental_data[\"plan_id_short\"] == rate_summary[\"PlanId\"], how=\"left\")\n","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Drop unnecessary DataFrames\ndental_data.cache()\nrate_summary.unpersist()\nnetwork.unpersist()\n\n# Deduplicate the merged DataFrame\ndental_data = dental_data.dropDuplicates()\n\ndental_data.printSchema()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# How many plans offer dental coverage of some sort?\nunique_plan_count = dental_data.select(\"plan_id_short\").distinct().count()\nprint(\"Number of plans offering dental coverage of some sort:\", unique_plan_count)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# How many providers do you have to choose from for dental?\nunique_provider_count = dental_data.select(\"IssuerId\").distinct().count()\nprint(\"Number of dental providers to choose from:\", unique_provider_count)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Where is dental available?\nstate_counts = dental_data.groupBy(\"StateCode\").count()\nstate_counts.show()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# How many are dental-only?\ndental_only_count = dental_data.filter(dental_data[\"DentalOnlyPlan\"] == \"Yes\").count()\n\nprint(\"Number of dental-only plans:\", dental_only_count)","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"spark.stop()","metadata":{"editable":false,"trusted":true},"execution_count":null,"outputs":[]}]}