From c62986de78046991447e12db0b16446744b8e89e Mon Sep 17 00:00:00 2001 From: Tom Jemmett Date: Tue, 22 Sep 2026 14:56:52 +0100 Subject: [PATCH 1/2] move demographic/birth factors into inputs data pipeline * creates the population data as tables in the reference layer * extracts these tables in the inputs-data layer * removes these extracts from the model data layer --- .../nhp_data-extract_nhp_for_containers.yaml | 36 +----- .../nhp_data-inputs-provider.yaml | 11 ++ .../nhp_data-reference_data.yaml | 11 ++ pyproject.toml | 4 +- src/nhp/data/aggregated_data/ecds.py | 8 +- src/nhp/data/aggregated_data/outpatients.py | 8 +- src/nhp/data/get_spark.py | 15 ++- src/nhp/data/inputs_data/demographics.py | 62 ++++++++++ src/nhp/data/model_data/birth_factors.py | 53 --------- .../data/model_data/demographic_factors.py | 50 --------- .../model_data/generate_synthetic_data.py | 22 ---- .../generate_icb_gams.py | 33 +++++- .../generate_provider_gams.py | 8 +- src/nhp/data/model_data/helpers.py | 76 ------------- src/nhp/data/model_data/ip_tpmas.py | 10 +- src/nhp/data/raw_data/aae.py | 4 +- src/nhp/data/raw_data/inpatients.py | 4 +- src/nhp/data/raw_data/outpatients.py | 4 +- .../data/raw_data/outpatients_mitigators.py | 4 +- src/nhp/data/reference/population_provider.py | 106 ++++++++++++++++++ src/nhp/data/table_names/mlcsu.py | 2 + src/nhp/data/table_names/table_names.py | 2 + src/nhp/data/table_names/udal.py | 2 + 23 files changed, 267 insertions(+), 268 deletions(-) create mode 100644 src/nhp/data/inputs_data/demographics.py delete mode 100644 src/nhp/data/model_data/birth_factors.py delete mode 100644 src/nhp/data/model_data/demographic_factors.py create mode 100644 src/nhp/data/reference/population_provider.py diff --git a/databricks_workflows/nhp_data-extract_nhp_for_containers.yaml b/databricks_workflows/nhp_data-extract_nhp_for_containers.yaml index 5641efe9..4f43bc9c 100644 --- a/databricks_workflows/nhp_data-extract_nhp_for_containers.yaml +++ b/databricks_workflows/nhp_data-extract_nhp_for_containers.yaml @@ -81,43 +81,9 @@ resources: job_cluster_key: nhp_data libraries: - whl: ../dist/*.whl - - task_key: run_extract_demographic_factors_data - depends_on: - - task_key: run_extract_ecds_data - for_each_task: - inputs: "{{job.parameters.years}}" - task: - task_key: extract_demographic_factors_data - python_wheel_task: - package_name: nhp_data - entry_point: model_data-demographic_factors - parameters: - - "{{job.parameters.data_version}}" - - "{{input}}" - - "{{job.parameters.projection_year}}" - job_cluster_key: nhp_data - libraries: - - whl: ../dist/*.whl - - task_key: run_extract_birth_factors_data - depends_on: - - task_key: run_extract_demographic_factors_data - for_each_task: - inputs: "{{job.parameters.years}}" - task: - task_key: extract_birth_factors_data - python_wheel_task: - package_name: nhp_data - entry_point: model_data-birth_factors - parameters: - - "{{job.parameters.data_version}}" - - "{{input}}" - - "{{job.parameters.projection_year}}" - job_cluster_key: nhp_data - libraries: - - whl: ../dist/*.whl - task_key: run_extract_inequalities_data depends_on: - - task_key: run_extract_birth_factors_data + - task_key: run_extract_ecds_data for_each_task: inputs: "{{job.parameters.years}}" task: diff --git a/databricks_workflows/nhp_data-inputs-provider.yaml b/databricks_workflows/nhp_data-inputs-provider.yaml index 79623114..8b54c791 100644 --- a/databricks_workflows/nhp_data-inputs-provider.yaml +++ b/databricks_workflows/nhp_data-inputs-provider.yaml @@ -86,6 +86,17 @@ resources: job_cluster_key: nhp_data libraries: - whl: ../dist/*.whl + - task_key: extract_demographics + depends_on: + - task_key: extract_inequalities + python_wheel_task: + package_name: nhp_data + entry_point: inputs_data-demographics + parameters: + - provider + job_cluster_key: nhp_data + libraries: + - whl: ../dist/*.whl job_clusters: - job_cluster_key: nhp_data new_cluster: diff --git a/databricks_workflows/nhp_data-reference_data.yaml b/databricks_workflows/nhp_data-reference_data.yaml index 6c0e936f..96c9c40d 100644 --- a/databricks_workflows/nhp_data-reference_data.yaml +++ b/databricks_workflows/nhp_data-reference_data.yaml @@ -83,6 +83,17 @@ resources: parameters: - "{{job.parameters.day_procedures_base_year}}" job_cluster_key: nhp_data + libraries: + - whl: ../dist/*.whl + - task_key: create_population_provider + depends_on: + - task_key: create_day_procedure_code_list + python_wheel_task: + package_name: nhp_data + entry_point: reference-population_provider + job_cluster_key: nhp_data + libraries: + - whl: ../dist/*.whl job_clusters: - job_cluster_key: nhp_data new_cluster: diff --git a/pyproject.toml b/pyproject.toml index 0ff62979..39c77e5a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,6 +74,7 @@ inputs_data-expat_repat = "nhp.data.inputs_data.expat_repat:main" inputs_data-procedures = "nhp.data.inputs_data.procedures:main" inputs_data-rates = "nhp.data.inputs_data.rates:main" inputs_data-inequalities = "nhp.data.inputs_data.inequalities:main" +inputs_data-demographics = "nhp.data.inputs_data.demographics:main" raw_data-aae = "nhp.data.raw_data.aae:main" raw_data-ecds = "nhp.data.raw_data.ecds:main" @@ -90,8 +91,6 @@ model_data-health_status_adjustment-generate_provider_gams = "nhp.data.model_dat model_data-health_status_adjustment-generate_icb_gams = "nhp.data.model_data.health_status_adjustment.generate_icb_gams:main" model_data-health_status_adjustment-generate_national_gams = "nhp.data.model_data.health_status_adjustment.generate_national_gams:main" model_data-aae = "nhp.data.model_data.aae:main" -model_data-birth_factors = "nhp.data.model_data.birth_factors:main" -model_data-demographic_factors = "nhp.data.model_data.demographic_factors:main" model_data-inequalities = "nhp.data.model_data.inequalities:main" model_data-ip = "nhp.data.model_data.ip:main" model_data-ip-tpmas = "nhp.data.model_data.ip_tpmas:main" @@ -108,5 +107,6 @@ reference-lsoa_lookups = "nhp.data.reference.lsoa_lookups:main" reference-population_by_lsoa21 = "nhp.data.reference.population_by_lsoa21:main" reference-provider_catchments = "nhp.data.reference.provider_catchments:main" reference-icb_catchments = "nhp.data.reference.icb_catchments:main" +reference-population_provider = "nhp.data.reference.population_provider:main" extract_data = "nhp.data.extract_data:main" diff --git a/src/nhp/data/aggregated_data/ecds.py b/src/nhp/data/aggregated_data/ecds.py index fff6ad76..409c0d0a 100644 --- a/src/nhp/data/aggregated_data/ecds.py +++ b/src/nhp/data/aggregated_data/ecds.py @@ -3,7 +3,7 @@ from pyspark.sql import DataFrame, SparkSession from pyspark.sql import functions as F -from nhp.data.get_spark import get_spark +from nhp.data.get_spark import PartitionOverwriteMode, get_spark from nhp.data.table_names import table_names @@ -42,7 +42,7 @@ def get_ecds_data(spark: SparkSession) -> DataFrame: ) -def generate_ecds_data(spark: SparkSession, ecds: DataFrame) -> None: +def generate_ecds_data(ecds: DataFrame) -> None: """Generate ECDS Data""" ( ecds.withColumn("index", F.expr("uuid()")) @@ -55,6 +55,6 @@ def generate_ecds_data(spark: SparkSession, ecds: DataFrame) -> None: def main() -> None: """main method""" - spark = get_spark() + spark = get_spark(PartitionOverwriteMode.STATIC) ecds = get_ecds_data(spark) - generate_ecds_data(spark, ecds) + generate_ecds_data(ecds) diff --git a/src/nhp/data/aggregated_data/outpatients.py b/src/nhp/data/aggregated_data/outpatients.py index 9a06d83f..bbb2ccb7 100644 --- a/src/nhp/data/aggregated_data/outpatients.py +++ b/src/nhp/data/aggregated_data/outpatients.py @@ -3,7 +3,7 @@ from pyspark.sql import DataFrame, SparkSession from pyspark.sql import functions as F -from nhp.data.get_spark import get_spark +from nhp.data.get_spark import PartitionOverwriteMode, get_spark from nhp.data.table_names import table_names @@ -45,7 +45,7 @@ def get_outpatients_data(spark: SparkSession) -> DataFrame: ) -def generate_outpatients_data(spark: SparkSession, opa: DataFrame) -> None: +def generate_outpatients_data(opa: DataFrame) -> None: """Generate Outpatients Data""" ( opa.write.partitionBy("fyear", "provider") @@ -57,6 +57,6 @@ def generate_outpatients_data(spark: SparkSession, opa: DataFrame) -> None: def main() -> None: """main method""" - spark = get_spark() + spark = get_spark(PartitionOverwriteMode.STATIC) opa = get_outpatients_data(spark) - generate_outpatients_data(spark, opa) + generate_outpatients_data(opa) diff --git a/src/nhp/data/get_spark.py b/src/nhp/data/get_spark.py index a74192d8..627cef84 100644 --- a/src/nhp/data/get_spark.py +++ b/src/nhp/data/get_spark.py @@ -1,9 +1,18 @@ """Helper method to get the spark session.""" +import enum + from pyspark.sql import SparkSession -def get_spark() -> SparkSession: +class PartitionOverwriteMode(enum.Enum): + DYNAMIC = "dynamic" + STATIC = "static" + + +def get_spark( + partition_overwrite_mode: PartitionOverwriteMode = PartitionOverwriteMode.DYNAMIC, +) -> SparkSession: """Get spark session :return: get the spark session to use @@ -14,5 +23,7 @@ def get_spark() -> SparkSession: from databricks.connect import DatabricksSession spark = DatabricksSession.builder.getOrCreate() - spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic") + spark.conf.set( + "spark.sql.sources.partitionOverwriteMode", partition_overwrite_mode.value + ) return spark diff --git a/src/nhp/data/inputs_data/demographics.py b/src/nhp/data/inputs_data/demographics.py new file mode 100644 index 00000000..d08fd6c3 --- /dev/null +++ b/src/nhp/data/inputs_data/demographics.py @@ -0,0 +1,62 @@ +"""Demographics/Birth/Catchments""" + +import sys + +import pyspark.sql.functions as F +from pyspark.sql import SparkSession + +from nhp.data.get_spark import get_spark +from nhp.data.inputs_data.save_parquet import save_parquet +from nhp.data.table_names import table_names + + +def extract_demographic_factors(path: str, spark: SparkSession) -> None: + """Extract Demographic Factors data + + :param path: the path to save the extracted data + :type path: str + :param spark: the spark session to use + :type spark: SparkSession + """ + df = spark.read.table(table_names.reference_population_provider_demographics) + save_parquet(df, f"{path}/demographic_factors") + + +def extract_birth_factors(path: str, spark: SparkSession) -> None: + """Extract Birth Factors data + + :param path: the path to save the extracted data + :type path: str + :param spark: the spark session to use + :type spark: SparkSession + """ + df = spark.read.table(table_names.reference_population_provider_births) + save_parquet(df, f"{path}/birth_factors") + + +def extract_provider_catchments(path: str, spark: SparkSession) -> None: + """Extract Provider Catchments data + + :param path: the path to save the extracted data + :type path: str + :param spark: the spark session to use + :type spark: SparkSession + """ + df = ( + spark.read.table(table_names.reference_provider_lad23_splits) + .filter(F.col("fyear") >= 202324) + .orderBy("fyear", "provider", "lad23cd", "sex", "age") + ) + save_parquet(df, f"{path}/provider_catchments") + + +def main() -> None: + geography_column = sys.argv[1] + assert geography_column in ["provider", "lad23cd"], "invalid geography_column" + + path = f"{table_names.inputs_save_path}/{geography_column}" + + spark = get_spark() + extract_demographic_factors(path, spark) + extract_birth_factors(path, spark) + extract_provider_catchments(path, spark) diff --git a/src/nhp/data/model_data/birth_factors.py b/src/nhp/data/model_data/birth_factors.py deleted file mode 100644 index 7a25a144..00000000 --- a/src/nhp/data/model_data/birth_factors.py +++ /dev/null @@ -1,53 +0,0 @@ -"""Extract birth factors data for model""" - -import sys - -import pyspark.sql.functions as F -from pyspark.sql import DataFrame, SparkSession - -from nhp.data.get_spark import get_spark -from nhp.data.model_data.helpers import ( - DEMOGRAPHICS_MAX_YEAR, - DEMOGRAPHICS_MIN_YEAR, - create_provider_population_projections, - extract, -) -from nhp.data.table_names import table_names - - -@extract("birth_factors") -def extract_birth_factors( - save_path: str, fyear: int, spark: SparkSession, projection_year: int -) -> DataFrame: - """Extract Birth Factors data - - :param save_path: where to save the parquet files - :type save_path: str - :param fyear: what year to extract - :type fyear: int - :param projection_year: the year for which to project the population - :type projection_year: int - :param spark: the spark session to use - :type spark: SparkSession - """ - births = ( - spark.read.table(table_names.population_projections_births) - .withColumn("sex", F.lit(2)) - .filter(F.col("year").between(DEMOGRAPHICS_MIN_YEAR, DEMOGRAPHICS_MAX_YEAR)) - ) - - return create_provider_population_projections( - spark, births, fyear, projection_year - ).withColumn("fyear", F.lit(fyear // 100)) - - -def main(): - data_version = sys.argv[1] - save_path = f"{table_names.model_data_path}/{data_version}" - - fyear = int(sys.argv[2]) - projection_year = int(sys.argv[3]) - - spark = get_spark() - - extract_birth_factors(save_path, fyear, spark, projection_year) diff --git a/src/nhp/data/model_data/demographic_factors.py b/src/nhp/data/model_data/demographic_factors.py deleted file mode 100644 index dcfe77fa..00000000 --- a/src/nhp/data/model_data/demographic_factors.py +++ /dev/null @@ -1,50 +0,0 @@ -"""Extract demographic factors data for model""" - -import sys - -import pyspark.sql.functions as F -from pyspark.sql import DataFrame, SparkSession - -from nhp.data.get_spark import get_spark -from nhp.data.model_data.helpers import ( - DEMOGRAPHICS_MAX_YEAR, - DEMOGRAPHICS_MIN_YEAR, - create_provider_population_projections, - extract, -) -from nhp.data.table_names import table_names - - -@extract("demographic_factors") -def extract_demographics( - save_path: str, fyear: int, spark: SparkSession, projection_year: int -) -> DataFrame: - """Extract Demographic Factors data - - :param save_path: where to save the parquet files - :type save_path: str - :param fyear: what year to extract - :type fyear: int - :param spark: the spark session to use - :type spark: SparkSession - :param projection_year: the year for which to project the population - :type projection_year: int - """ - demographics = spark.read.table( - table_names.population_projections_demographics - ).filter(F.col("year").between(DEMOGRAPHICS_MIN_YEAR, DEMOGRAPHICS_MAX_YEAR)) - - return create_provider_population_projections( - spark, demographics, fyear, projection_year - ).withColumn("fyear", F.lit(fyear // 100)) - - -def main(): - data_version = sys.argv[1] - save_path = f"{table_names.model_data_path}/{data_version}" - fyear = int(sys.argv[2]) - projection_year = int(sys.argv[3]) - - spark = get_spark() - - extract_demographics(save_path, fyear, spark, projection_year) diff --git a/src/nhp/data/model_data/generate_synthetic_data.py b/src/nhp/data/model_data/generate_synthetic_data.py index d3459b65..bb705d27 100644 --- a/src/nhp/data/model_data/generate_synthetic_data.py +++ b/src/nhp/data/model_data/generate_synthetic_data.py @@ -75,8 +75,6 @@ def generate(self) -> None: self._inequalities() self._aae() self._op() - self._birth_factors() - self._demographic_factors() self._hsa_activity_tables() @property @@ -392,26 +390,6 @@ def _op(self, df: DataFrame) -> pd.DataFrame: return op - @generate_data("birth_factors") - def _birth_factors(self, df: DataFrame) -> pd.DataFrame: - return ( - df.drop("dataset") - .filter(~F.col("variant").startswith("custom_projection_")) - .toPandas() - .groupby(["variant", "sex", "age"], as_index=False) - .mean() - ) - - @generate_data("demographic_factors") - def _demographic_factors(self, df: DataFrame) -> pd.DataFrame: - return ( - df.drop("dataset") - .filter(~F.col("variant").startswith("custom_projection_")) - .toPandas() - .groupby(["variant", "sex", "age"], as_index=False) - .mean() - ) - @generate_data("hsa_activity_tables") def _hsa_activity_tables(self, df: DataFrame) -> pd.DataFrame: return ( diff --git a/src/nhp/data/model_data/health_status_adjustment/generate_icb_gams.py b/src/nhp/data/model_data/health_status_adjustment/generate_icb_gams.py index c390c34c..8dd04023 100644 --- a/src/nhp/data/model_data/health_status_adjustment/generate_icb_gams.py +++ b/src/nhp/data/model_data/health_status_adjustment/generate_icb_gams.py @@ -12,10 +12,37 @@ from pyspark.sql import DataFrame, SparkSession, Window from nhp.data.get_spark import get_spark -from nhp.data.model_data.helpers import create_icb_population_projections from nhp.data.table_names import table_names +def _create_icb_population_projections( + spark: SparkSession, projection_year: int = 2022 +) -> DataFrame: + catchments = spark.read.table(table_names.reference_icb_catchments) + + projections_to_include = [ + "migration_category", + "var_proj_5_year_migration", + "var_proj_10_year_migration", + "var_proj_high_intl_migration", + "var_proj_low_intl_migration", + "var_proj_zero_net_migration", + ] + + return ( + spark.read.table(table_names.population_projections_demographics) + .filter(F.col("projection_year") == projection_year) + .filter(F.col("projection").isin(projections_to_include)) + .join(catchments, ["area_code"]) + .withColumnRenamed("projection", "variant") + .withColumnRenamed("provider", "dataset") + .groupBy("icb", "variant", "age", "sex") + .pivot("year") + .agg(F.sum(F.col("value") * F.col("pcnt"))) + .orderBy("icb", "variant", "age", "sex") + ) + + def _get_data(spark: SparkSession, save_path: str, years: list[int]) -> DataFrame: dfr = ( reduce( @@ -46,9 +73,7 @@ def _get_data(spark: SparkSession, save_path: str, years: list[int]) -> DataFram # load the demographics data demog = ( - create_icb_population_projections( - spark, spark.read.table(table_names.population_projections_demographics) - ) + _create_icb_population_projections(spark) .filter(F.col("variant") == "migration_category") .filter(F.col("age") >= 18) .selectExpr( diff --git a/src/nhp/data/model_data/health_status_adjustment/generate_provider_gams.py b/src/nhp/data/model_data/health_status_adjustment/generate_provider_gams.py index f2433b7a..02fffa05 100644 --- a/src/nhp/data/model_data/health_status_adjustment/generate_provider_gams.py +++ b/src/nhp/data/model_data/health_status_adjustment/generate_provider_gams.py @@ -51,19 +51,21 @@ def _get_data(spark: SparkSession, save_path: str, years: list[int]) -> DataFram # load the demographics data demog = ( - spark.read.parquet(f"{save_path}/demographic_factors/") + spark.read.parquet( + f"{table_names.inputs_save_path}/provider/demographic_factors.parquet" + ) .filter(F.col("variant") == "migration_category") .filter(F.col("age") >= 18) .selectExpr( "age", "sex", - "dataset", + "provider as dataset", "fyear as base_fyear", f"stack({len(years)}, " + ", ".join([f"'{y}', `{y}`" for y in years]) + ") as (fyear, pop)", ) - .filter(F.col("base_fyear") == F.col("fyear")) + .filter(F.col("base_fyear").cast("string").substr(0, 4) == F.col("fyear")) .drop("base_fyear") .withColumn("fyear", F.col("fyear").cast("int")) .join( diff --git a/src/nhp/data/model_data/helpers.py b/src/nhp/data/model_data/helpers.py index 6bf43da5..286ecabd 100644 --- a/src/nhp/data/model_data/helpers.py +++ b/src/nhp/data/model_data/helpers.py @@ -6,82 +6,6 @@ import pyspark.sql.functions as F from pyspark.sql import DataFrame, SparkSession -from nhp.data.reference.lsoa_lookups import get_lad22_to_lad23_lookup -from nhp.data.table_names import table_names - -# what years should we support in the extract? -DEMOGRAPHICS_MIN_YEAR, DEMOGRAPHICS_MAX_YEAR = 2023, 2047 - - -def create_provider_population_projections( - spark: SparkSession, df: DataFrame, fyear: int, projection_year: int = 2022 -) -> DataFrame: - providers = ( - spark.read.table(table_names.reference_ods_trusts) - .filter(F.col("org_type").startswith("ACUTE")) - .select(F.col("org_to").alias("provider")) - .distinct() - ) - - catchments = ( - spark.read.table(table_names.reference_provider_lad23_splits) - .filter(F.col("fyear") == fyear) - .drop("fyear") - .join(providers, "provider", how="semi") - ) - - projections_to_include = [ - "migration_category", - "var_proj_5_year_migration", - "var_proj_10_year_migration", - "var_proj_high_intl_migration", - "var_proj_low_intl_migration", - "var_proj_zero_net_migration", - ] - - lad22_to_lad23 = get_lad22_to_lad23_lookup(spark) - - return ( - df.filter(F.col("projection_year") == projection_year) - .filter(F.col("projection").isin(projections_to_include)) - .withColumnRenamed("area_code", "lad22cd") - .join(lad22_to_lad23, "lad22cd") - .join(catchments, ["lad23cd", "age", "sex"]) - .withColumnRenamed("projection", "variant") - .withColumnRenamed("provider", "dataset") - .groupBy("dataset", "variant", "age", "sex") - .pivot("year") - .agg(F.sum(F.col("value") * F.col("lad23_pcnt"))) - .orderBy("dataset", "variant", "age", "sex") - ) - - -def create_icb_population_projections( - spark: SparkSession, df: DataFrame, projection_year: int = 2022 -) -> DataFrame: - catchments = spark.read.table(table_names.reference_icb_catchments) - - projections_to_include = [ - "migration_category", - "var_proj_5_year_migration", - "var_proj_10_year_migration", - "var_proj_high_intl_migration", - "var_proj_low_intl_migration", - "var_proj_zero_net_migration", - ] - - return ( - df.filter(F.col("projection_year") == projection_year) - .filter(F.col("projection").isin(projections_to_include)) - .join(catchments, ["area_code"]) - .withColumnRenamed("projection", "variant") - .withColumnRenamed("provider", "dataset") - .groupBy("icb", "variant", "age", "sex") - .pivot("year") - .agg(F.sum(F.col("value") * F.col("pcnt"))) - .orderBy("icb", "variant", "age", "sex") - ) - def check_extract_for_nulls( df: DataFrame, exclude_cols: set[str] | None = None diff --git a/src/nhp/data/model_data/ip_tpmas.py b/src/nhp/data/model_data/ip_tpmas.py index 449d6d6b..c299720b 100644 --- a/src/nhp/data/model_data/ip_tpmas.py +++ b/src/nhp/data/model_data/ip_tpmas.py @@ -43,11 +43,11 @@ def extract_ip_activity_avoidance_tpmas( return _extract_ip_tpmas("activity_avoidance", save_path, fyear, spark) -@extract("ip_efficiency_strategies") -def extract_ip_efficiency_tpmas( +@extract("ip_efficiencies_strategies") +def extract_ip_efficiencies_tpmas( save_path: str, fyear: int, spark: SparkSession ) -> DataFrame: - """Extract IP efficiency TPMAs data + """Extract IP efficiencies TPMAs data :param save_path: where to save the parquet files :type save_path: str @@ -56,7 +56,7 @@ def extract_ip_efficiency_tpmas( :param spark: the spark session to use :type spark: SparkSession """ - return _extract_ip_tpmas("efficiencies", save_path, fyear, spark) + return _extract_ip_tpmas("efficiency", save_path, fyear, spark) def main(): @@ -67,4 +67,4 @@ def main(): spark = get_spark() extract_ip_activity_avoidance_tpmas(save_path, fyear, spark) - extract_ip_efficiency_tpmas(save_path, fyear, spark) + extract_ip_efficiencies_tpmas(save_path, fyear, spark) diff --git a/src/nhp/data/raw_data/aae.py b/src/nhp/data/raw_data/aae.py index b529b433..84b8fd9c 100644 --- a/src/nhp/data/raw_data/aae.py +++ b/src/nhp/data/raw_data/aae.py @@ -3,7 +3,7 @@ import pyspark.sql.functions as F from pyspark.sql import DataFrame, SparkSession -from nhp.data.get_spark import get_spark +from nhp.data.get_spark import PartitionOverwriteMode, get_spark from nhp.data.nhp_datasets.icbs import add_main_icb, icb_mapping from nhp.data.nhp_datasets.local_authorities import lsoa11_to_lad23 from nhp.data.nhp_datasets.providers import read_data_with_provider @@ -216,5 +216,5 @@ def generate_aae_data(spark: SparkSession) -> None: def main() -> None: """main method""" - spark = get_spark() + spark = get_spark(PartitionOverwriteMode.STATIC) generate_aae_data(spark) diff --git a/src/nhp/data/raw_data/inpatients.py b/src/nhp/data/raw_data/inpatients.py index f500fab0..cb5ecb73 100644 --- a/src/nhp/data/raw_data/inpatients.py +++ b/src/nhp/data/raw_data/inpatients.py @@ -3,7 +3,7 @@ from pyspark.sql import DataFrame, SparkSession from pyspark.sql import functions as F -from nhp.data.get_spark import get_spark +from nhp.data.get_spark import PartitionOverwriteMode, get_spark from nhp.data.nhp_datasets.apc import apc_primary_procedures, hes_apc from nhp.data.nhp_datasets.icbs import add_main_icb from nhp.data.raw_data.helpers import ( @@ -181,5 +181,5 @@ def generate_inpatients_data(spark: SparkSession) -> None: def main() -> None: """main method""" - spark = get_spark() + spark = get_spark(partition_overwrite_mode=PartitionOverwriteMode.STATIC) generate_inpatients_data(spark) diff --git a/src/nhp/data/raw_data/outpatients.py b/src/nhp/data/raw_data/outpatients.py index b10565dc..58910dfc 100644 --- a/src/nhp/data/raw_data/outpatients.py +++ b/src/nhp/data/raw_data/outpatients.py @@ -3,7 +3,7 @@ from pyspark.sql import DataFrame, SparkSession from pyspark.sql import functions as F -from nhp.data.get_spark import get_spark +from nhp.data.get_spark import PartitionOverwriteMode, get_spark from nhp.data.nhp_datasets.icbs import add_main_icb, icb_mapping from nhp.data.nhp_datasets.local_authorities import lsoa11_to_lad23 from nhp.data.nhp_datasets.providers import read_data_with_provider @@ -176,5 +176,5 @@ def generate_outpatients_data(spark: SparkSession) -> None: def main() -> None: """main method""" - spark = get_spark() + spark = get_spark(PartitionOverwriteMode.STATIC) generate_outpatients_data(spark) diff --git a/src/nhp/data/raw_data/outpatients_mitigators.py b/src/nhp/data/raw_data/outpatients_mitigators.py index 0cec4d1d..81e0bab3 100644 --- a/src/nhp/data/raw_data/outpatients_mitigators.py +++ b/src/nhp/data/raw_data/outpatients_mitigators.py @@ -5,7 +5,7 @@ from pyspark.sql import DataFrame, SparkSession from pyspark.sql import functions as F -from nhp.data.get_spark import get_spark +from nhp.data.get_spark import PartitionOverwriteMode, get_spark from nhp.data.table_names import table_names @@ -58,5 +58,5 @@ def generate_outpatients_mitigators(spark: SparkSession) -> None: def main() -> None: """main method""" - spark = get_spark() + spark = get_spark(PartitionOverwriteMode.STATIC) generate_outpatients_mitigators(spark) diff --git a/src/nhp/data/reference/population_provider.py b/src/nhp/data/reference/population_provider.py new file mode 100644 index 00000000..5cdcec6e --- /dev/null +++ b/src/nhp/data/reference/population_provider.py @@ -0,0 +1,106 @@ +"""Helper methods/tables""" + +import pyspark.sql.functions as F +from pyspark.sql import DataFrame, SparkSession + +from nhp.data.get_spark import get_spark +from nhp.data.reference.lsoa_lookups import get_lad22_to_lad23_lookup +from nhp.data.table_names import table_names + +# what years should we support in the extract? +DEMOGRAPHICS_MIN_YEAR, DEMOGRAPHICS_MAX_YEAR = 2023, 2047 +PROJECTIONS_TO_INCLUDE = [ + "migration_category", + "var_proj_5_year_migration", + "var_proj_10_year_migration", + "var_proj_high_intl_migration", + "var_proj_low_intl_migration", + "var_proj_zero_net_migration", +] +PROJECTION_YEAR = 2022 + + +def create_provider_demographics( + spark: SparkSession, table_from: str, table_to: str +) -> None: + providers = ( + spark.read.table(table_names.reference_ods_trusts) + .filter(F.col("org_type").startswith("ACUTE")) + .select(F.col("org_to").alias("provider")) + .distinct() + ) + + min_fyear = DEMOGRAPHICS_MIN_YEAR * 100 + (DEMOGRAPHICS_MIN_YEAR + 1) % 100 + catchments = ( + spark.read.table(table_names.reference_provider_lad23_splits) + .filter(F.col("fyear") >= min_fyear) + .join(providers, "provider", how="semi") + ) + + lad22_to_lad23 = get_lad22_to_lad23_lookup(spark) + + df = spark.read.table(table_from) + + if table_from == table_names.population_projections_births: + df = df.withColumn("sex", F.lit(2)) + + df = ( + df.filter(F.col("year").between(DEMOGRAPHICS_MIN_YEAR, DEMOGRAPHICS_MAX_YEAR)) + .filter(F.col("projection_year") == PROJECTION_YEAR) + .filter(F.col("projection").isin(PROJECTIONS_TO_INCLUDE)) + .withColumnRenamed("projection", "variant") + .withColumnRenamed("area_code", "lad22cd") + .join(lad22_to_lad23, "lad22cd") + .join(catchments, ["lad23cd", "age", "sex"]) + .groupBy("fyear", "provider", "variant", "sex", "age") + .pivot("year") + .agg(F.sum(F.col("value") * F.col("lad23_pcnt"))) + .orderBy("fyear", "provider", "variant", "sex", "age") + ) + df.repartition(1).write.mode("overwrite").partitionBy( + "fyear", "provider", "variant" + ).saveAsTable(table_to) + + +def create_demographic_factors(spark: SparkSession) -> None: + """Extract Demographic Factors data + + :param spark: the spark session to use + :type spark: SparkSession + """ + create_provider_demographics( + spark, + table_names.population_projections_demographics, + table_names.reference_population_provider_demographics, + ) + + +def create_birth_factors(spark: SparkSession) -> None: + """Extract Birth Factors data + + :param spark: the spark session to use + :type spark: SparkSession + """ + create_provider_demographics( + spark, + table_names.population_projections_births, + table_names.reference_population_provider_births, + ) + + +def main(): + """Main entry point for generating provider catchment population tables. + + This function creates the population by provider table, which includes calculating + provider-to-LAD23 splits based on APC data and applying those splits to population + estimates. The get_pop_by_provider() call will trigger creation of all dependent + tables if they don't already exist. + + Tables generated: + - Provider to LAD23 splits (if not exists) + - Population by provider (if not exists) + """ + spark = get_spark() + + create_demographic_factors(spark) + create_birth_factors(spark) diff --git a/src/nhp/data/table_names/mlcsu.py b/src/nhp/data/table_names/mlcsu.py index cc98fa84..45d1571b 100644 --- a/src/nhp/data/table_names/mlcsu.py +++ b/src/nhp/data/table_names/mlcsu.py @@ -50,6 +50,8 @@ reference_lsoa11_to_lad23="nhp.reference.lsoa11_to_lad23", reference_lad22_to_lad23="nhp.reference.lad22_to_lad23", reference_theatre_times="nhp.reference.theatre_times", + reference_population_provider_births="unhp.population_projections.provider_births", + reference_population_provider_demographics="unhp.population_projections.provider_demographics", # --- reference_day_procedures_code_list="/Volumes/nhp/reference/files/day_procedures.json", reference_frailty_risk_scores="/Volumes/nhp/reference/files/frailty_risk_scores.csv", diff --git a/src/nhp/data/table_names/table_names.py b/src/nhp/data/table_names/table_names.py index 3528a51c..a9d5fab2 100644 --- a/src/nhp/data/table_names/table_names.py +++ b/src/nhp/data/table_names/table_names.py @@ -52,6 +52,8 @@ class TableNames: reference_pop_by_lad23: str reference_lad22_to_lad23: str reference_theatre_times: str + reference_population_provider_births: str + reference_population_provider_demographics: str # TODO: convert these to tables reference_day_procedures_code_list: str reference_frailty_risk_scores: str diff --git a/src/nhp/data/table_names/udal.py b/src/nhp/data/table_names/udal.py index c65e18b3..f7040668 100644 --- a/src/nhp/data/table_names/udal.py +++ b/src/nhp/data/table_names/udal.py @@ -50,6 +50,8 @@ reference_lsoa11_to_lad23="udal_lake_mart.newhospitalprogramme.reference_lsoa11_to_lad23", reference_lad22_to_lad23="udal_lake_mart.newhospitalprogramme.reference_lad22_to_lad23", reference_theatre_times="udal_lake_mart.newhospitalprogramme.reference_theatre_times", + reference_population_provider_births="udal_lake_mart.newhospitalprogramme.population_provider_births", + reference_population_provider_demographics="udal_lake_mart.newhospitalprogramme.population_provider_demographics", # --- reference_day_procedures_code_list="/Volumes/udal_lake_mart/newhospitalprogramme/files/reference/day_procedures.json", reference_frailty_risk_scores="/Volumes/udal_lake_mart/newhospitalprogramme/files/reference/frailty_risk_scores.csv", From 52ecc2d070e2a116a74e0e4c6be55422c81e4e9a Mon Sep 17 00:00:00 2001 From: Tom Jemmett Date: Tue, 13 Oct 2026 13:55:22 +0100 Subject: [PATCH 2/2] fix lint issue --- src/nhp/data/reference/population_provider.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/nhp/data/reference/population_provider.py b/src/nhp/data/reference/population_provider.py index 5cdcec6e..0a1756fd 100644 --- a/src/nhp/data/reference/population_provider.py +++ b/src/nhp/data/reference/population_provider.py @@ -1,7 +1,7 @@ """Helper methods/tables""" import pyspark.sql.functions as F -from pyspark.sql import DataFrame, SparkSession +from pyspark.sql import SparkSession from nhp.data.get_spark import get_spark from nhp.data.reference.lsoa_lookups import get_lad22_to_lad23_lookup