diff --git a/data/ukb_simulated_data/example_ukb_rap_convert.ipynb b/data/ukb_simulated_data/example_ukb_rap_convert.ipynb index 57a5257..2d54034 100644 --- a/data/ukb_simulated_data/example_ukb_rap_convert.ipynb +++ b/data/ukb_simulated_data/example_ukb_rap_convert.ipynb @@ -2,213 +2,270 @@ "cells": [ { "cell_type": "markdown", - "id": "c6606cb2", + "id": "f5d2eb46-bb37-4d1d-a93f-eda0d307393e", "metadata": {}, "source": [ - "# General example for converting UK Biobank data into the research environemnt (RAP) to DELPHI format\n" + "## File paths and train,validation split" + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "id": "8319ff7b-db17-4eea-8f86-6e0fa47c4bb8", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "import pandas as pd\n", + "import tqdm\n", + "import numpy as np\n", + "import pyspark\n", + "import dxpy\n", + "import dxdata\n", + "import pandas as pd\n", + "import numpy as np\n", + "\n", + "labels_file = 'labels.csv'\n", + "ukb_field_to_icd10_map_file = 'icd10_codes_mod.tsv'\n", + "ubk_basket_tab_file = 'ukb_basket.tab' # file path to ukb basket download .tab format\n", + "train_proportion = 0.8 # proportion of full data set to use for training (the rest will be used for validation)\n", + "output_prefix = 'ukb_real'\n" ] }, { "cell_type": "markdown", - "id": "ba4bfe9e", + "id": "f7d089aa-56de-42ae-86b6-5320675c5729", "metadata": {}, "source": [ - "Notes:\n", - " - This is setup to run as a notebook in a spark server job\n", - " - it needs access to a dataset record (which you may need to explicitly specify for your project)\n", - " - and also a cohort which is here refered to as \"full_cohort\" this may differ from your project\n", - " - the token labels to ukb field ids should be modified to suit your needs\n", - " - your should change the token ids to be integer and create a token_id to field_id (or disease icd10 name)\n", - " - this output a file with all individuals included - you will want to split this into \"train.bin\" and \"val.bin\" " + "## Read icd10 mapping file and defined index label link" ] }, { "cell_type": "code", - "execution_count": null, - "id": "3c85eb2e", + "execution_count": 2, + "id": "b75654c9-c8ef-42f9-8569-97a97378dca5", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "icdict ={}\n", + "icdcodes = []\n", + "with open(ukb_field_to_icd10_map_file,'r') as f:\n", + " for l in f:\n", + " lvals=l.strip().split()\n", + " icdict[\"p\" + lvals[0].split(\".\")[1]]=lvals[5]\n", + " icdcodes.append(lvals[5])\n", + "\n", + "i = -1\n", + "label_dict = {}\n", + "with open(labels_file,'r') as f:\n", + " for l in f:\n", + " label_dict[l.strip().split(' ')[0]]=i\n", + " i += 1\n", + "\n", + "# hard coded sex and dob\n", + "icdict['p31'] = \"sex\"\n", + "icdict['p34'] = \"YEAR\"\n", + "icdict['p52'] = \"MONTH\"\n", + "icdict['p40000_i0'] = \"Death\"\n", + "\n", + "# cancer fields\n", + "for j in range(17):\n", + " icdict['p40005_i'+str(j)] = \"cancer_date_\"+str(j)\n", + " icdict['p40006_i'+str(j)] = \"cancer_type_\"+str(j)\n", + "\n", + "# cancer hes fields \n", + "#for j in range(213):\n", + "# icdict['f.41270.0.'+str(j)] = \"hicd_\"+str(j)\n", + "# icdict['f.41280.0.'+str(j)] = \"hicd_date_\"+str(j)\n", + "\n", + "icdict['p53_i0'] = \"assessment_date\"\n", + "icdict['p21001_i0']=\"BMI\"\n", + "icdict['p1239_i0']=\"smoking\"\n", + "icdict['p1558_i0']=\"alcohol\"\n", + "\n", + "len_icd = len(icdcodes)\n", + "#icdcodes.extend(['Death','assessment_date']+['cancer_date_'+str(j) for j in range(17)]+['hicd_date_'+str(j) for j in range(213)])\n", + "icdcodes.extend(['Death','assessment_date']+['cancer_date_'+str(j) for j in range(17)])\n" + ] + }, + { + "cell_type": "markdown", + "id": "43fa0d3f-f04d-462f-a95d-709f2aeb7907", "metadata": {}, + "source": [ + "## Retrieve ukb fields from SQL database" + ] + }, + { + "cell_type": "code", + "execution_count": 3, + "id": "0763e6f1-fcb9-4e90-972b-654cf77c0755", + "metadata": { + "tags": [] + }, "outputs": [], "source": [ - "import dxdata\n", - "import pandas as pd\n", - "import numpy as np\n", - "from tqdm import tqdm\n", - "from pyspark.sql.functions import lit, udf, col\n", - "from pyspark.sql.types import DoubleType\n", - "from functools import reduce\n", - "import os\n", - "\n", - "def get_first_occ_fields(main_entity):\n", - " fo_fields = []\n", - " for field in main_entity.fields:\n", - " parts = field.name.split(\"_\")\n", - " if len(str(parts[0])) > 3:\n", - " field_num = int(parts[0][1:])\n", - " if (field_num >= 130000 and field_num <= 132604):\n", - " if field.title.startswith(\"Date\"):\n", - " fo_fields.append(field)\n", - " return fo_fields\n", - "\n", - "def compute_age_from_eid_and_event(eid, event_date):\n", - " dob = dob_lookup.get(eid)\n", - " if dob is None or event_date is None:\n", - " return None\n", - " try:\n", - " return (pd.to_datetime(event_date) - dob).days / 365.25\n", - " except Exception:\n", - " return None\n", - "\n", - "# Initialize dxdata engine\n", - "engine = dxdata.connect(dialect=\"hive+pyspark\")\n", - "\n", - "project = os.getenv('DX_PROJECT_CONTEXT_ID')\n", - "record = os.popen(\"dx find data --type Dataset --delimiter ',' | awk -F ',' '{print $5}'\").read().rstrip()\n", - "# find what is presumed to be the relevant dataset record\n", - "record = record.split('\\n')[0]\n", - "\n", - "DATASET_ID = project + \":\" + record\n", - "dataset = dxdata.load_dataset(id=DATASET_ID)\n", - "\n", - "# we retrieve the priamry entity from the dataset\n", - "main_entity = dataset.primary_entity\n", - "\n", - "# use cohort - change to whichever name:path you have for this object\n", - "cohort = dxdata.load_cohort(folder=\"/\", name=\"full_cohort\")\n", - "cohort_eids_df = engine.execute(cohort.sql)" + "def get_database_id():\n", + " dispensed_dataset = dxpy.find_one_data_object(\n", + " typename='Dataset', \n", + " name='app*.dataset', \n", + " folder='/', \n", + " name_mode='glob')\n", + " dispensed_dataset_id = dispensed_dataset['id']\n", + " return(dispensed_dataset_id)\n", + "\n", + "sc = pyspark.SparkContext()\n", + "spark = pyspark.sql.SparkSession(sc)\n", + "\n", + "dispensed_dataset_id = get_database_id()\n", + "\n", + "dataset = dxdata.load_dataset(id=dispensed_dataset_id)\n", + "participant = dataset['participant']" ] }, { "cell_type": "code", - "execution_count": null, - "id": "361f018d-fe1c-40ec-b2f4-d43f7c9887ca", + "execution_count": 4, + "id": "524398cb-132e-4c70-9293-d40d4997b002", "metadata": { "tags": [] }, "outputs": [], "source": [ - "# hard coded sex dob and basic demographic data\n", - "eid_f = main_entity.find_field(name=\"eid\")\n", - "sex_f = main_entity.find_field(title=\"Sex\")\n", - "year_f = main_entity.find_field(title=\"Year of birth\")\n", - "month_f = main_entity.find_field(title=\"Month of birth\")\n", - "death_f = dataset['death'].find_field(title=\"Date of death\")\n", - "assessment_f = main_entity[\"p53_i0\"]\n", - "bmi_f = main_entity[\"p21001_i0\"]\n", - "smoking_f = main_entity[\"p1239_i0\"]\n", - "alcohol_f = main_entity[\"p1558_i0\"]\n" + "fields_to_get = [\"eid\"] + list(icdict.keys())" ] }, { "cell_type": "code", "execution_count": 5, - "id": "004328b8-abfe-4e63-b704-87efbe7138cd", + "id": "a315ed54-f924-4337-8886-c5d82480e5b6", "metadata": { "tags": [] }, "outputs": [], "source": [ - "# collect the cancer code enteries\n", - "cancer_codes = {}\n", - "cancer_codes['type'] = []\n", - "cancer_codes['date'] = []\n", - "for i in range(22):\n", - " cancer_codes['type'].append(main_entity.find_field(name=\"p40006_i\" + str(i)))\n", - " cancer_codes['date'].append(main_entity.find_field(name=\"p40005_i\" + str(i)))\n" + "dd_sql = participant.retrieve_fields(names=fields_to_get, \n", + " coding_values = \"raw\",\n", + " engine=dxdata.connect(dialect=\"hive+pyspark\"))\n", + "dd = dd_sql.toPandas()" ] }, { "cell_type": "code", - "execution_count": null, - "id": "0381027b-de51-494e-aa12-d3a0c126d35f", + "execution_count": 6, + "id": "83d741c0-6bc4-43ff-9623-42a7ec625011", "metadata": { "tags": [] }, "outputs": [], "source": [ - "# Deal with first occrances and demographic data - this takes a little while\n", - "fo_fields = get_first_occ_fields(main_entity)\n", - "fields_to_get = [eid_f, sex_f, year_f, month_f, assessment_f, bmi_f, smoking_f, alcohol_f] + cancer_codes['type'] + cancer_codes['date'] + fo_fields + [death_f]\n", - "\n", - "df = main_entity.retrieve_fields(fields=fields_to_get, filter_sql=cohort.sql, engine=engine)\n", - "df1 = df.select(\"eid\", \"p31\",\"p34\",\"p52\",\"p53_i0\",\"p21001_i0\",\"p1239_i0\",\"p1558_i0\").toPandas()\n", - "\n", - "dobf1 = df1[['p34', 'p52']]\n", - "dobf1.columns = [\"YEAR\", \"MONTH\"]\n", - "df1['dob'] = pd.to_datetime(dobf1.assign(DAY=1))\n", - "df1['bmi_status'] = np.where(df1['p21001_i0']>28,5,np.where(df1['p21001_i0']>22,4,3))\n", - "df1['smoking_status'] = np.where(df1['p1239_i0']==1,8,np.where(df1['p1239_i0']==2,7,6))\n", - "df1['alcohol_status'] = np.where(df1['p1558_i0']==1,11,np.where(df1['p1558_i0'] < 4,10,9))" + "dd = dd.set_index(\"eid\")" ] }, { - "cell_type": "code", - "execution_count": null, - "id": "343290db-0f6c-4717-bc25-b821005c0c31", + "cell_type": "markdown", + "id": "09c99035-38a2-4ad5-9ae2-2768abc69d9d", "metadata": {}, - "outputs": [], "source": [ - "# Prepare a pandas dictionary for fast eid to dob lookup\n", - "dob_lookup = df1.set_index('eid')['dob'].to_dict()\n", - "age_event_udf = udf(compute_age_from_eid_and_event, DoubleType())" + "## format for delphi" ] }, { "cell_type": "code", - "execution_count": null, - "id": "d0a7b99c-7dd9-43c8-9766-b10256282117", - "metadata": {}, + "execution_count": 7, + "id": "b27ba642-d21b-4925-b4a6-c149eadff487", + "metadata": { + "tags": [] + }, "outputs": [], "source": [ - "# Remove all NULL enteries for each ICD10 code seperately and combine into an overall spark table \n", + "data_list = []\n", + "\n", + "dd = dd.rename(columns=icdict)\n", + "dd.dropna(subset=['sex'], inplace=True)\n", + "dd['sex'] += 1\n", + "dd = dd[[col for col in dd.columns if not col.startswith('f.')]]\n", + "dd['dob'] = pd.to_datetime(dd[['YEAR', 'MONTH']].assign(DAY=1))\n", + "dd[icdcodes] = dd[icdcodes].apply(pd.to_datetime, format=\"%Y-%m-%d\")\n", + "dd[icdcodes]=dd[icdcodes].sub(dd['dob'], axis=0)\n", + "dd[icdcodes]=dd[icdcodes].apply(lambda x : x.dt.days)\n", "\n", - "# deal with the tokens and dates \n", - "d_all = df.select(\"eid\", cancer_codes['date'][0].name, cancer_codes['type'][0].name).where(df[cancer_codes['date'][0].name].isNotNull())\n", - "d_all = d_all.withColumnRenamed(cancer_codes['date'][i].name, \"date\")\n", - "d_all = d_all.withColumnRenamed(cancer_codes['type'][i].name, \"token\")\n", - "d_all = d_all.withColumn(\"age\", age_event_udf(col(\"eid\"), col(\"date\")))\n", + "for col in icdcodes[:len_icd+1]:\n", + " X = dd[col].dropna().reset_index().to_numpy().astype(int)\n", + " data_list.append(np.hstack((X,label_dict[col]*np.ones([X.shape[0],1],X.dtype))))\n", + "\n", + "X = dd['sex'].reset_index().to_numpy().astype(int)\n", + "data_list.append(np.c_[X[:,0],np.zeros(X.shape[0]),X[:,1]].astype(int))\n", + "\n", + "for j in range(17):\n", + " dd_cancer = dd[['cancer_date_'+str(j),'cancer_type_'+str(j)]].dropna().reset_index()\n", + " if not dd_cancer.empty:\n", + " dd_cancer['cancer'] = dd_cancer['cancer_type_'+str(j)].str.slice(0,3)\n", + " dd_cancer['cancer_label'] = dd_cancer[\"cancer\"].map(label_dict)\n", + " data_list.append(dd_cancer[['eid','cancer_date_'+str(j),'cancer_label']].dropna().astype(int).to_numpy())\n", + "\n", + "#for j in range(213):\n", + "# dd_hicd = dd[['hicd_date_'+str(j),'hicd_'+str(j)]].dropna().reset_index()\n", + "# if not dd_hicd.empty:\n", + "# dd_hicd['hicd'] = dd_hicd['hicd_'+str(j)].str.slice(0,3)\n", + "# dd_hicd['hicd_label'] = dd_hicd[\"hicd\"].map(label_dict)\n", + "# data_list.append(dd_hicd[['f.eid','hicd_date_'+str(j),'hicd_label']].dropna().astype(int).to_numpy())\n", " \n", - "for i in tqdm(1, len(cancer_codes['date'])):\n", - " cf1 = df.select(\"eid\", cancer_codes['date'][i].name, cancer_codes['type'][i].name).where(df[cancer_codes['date'][i].name].isNotNull())\n", - " cf1 = cf1.withColumnRenamed(cancer_codes['date'][i].name, \"date\")\n", - " cf1 = cf1.withColumnRenamed(cancer_codes['type'][i].name, \"token\")\n", - " cf1 = cf1.withColumn(\"age\", age_event_udf(col(\"eid\"), col(\"date\")))\n", - " d_all = d_all.union(cf1)\n" + "dd_bmi = dd[['assessment_date','BMI']].dropna().reset_index()\n", + "dd_bmi['bmi_status'] = np.where(dd_bmi['BMI']>28,5,np.where(dd_bmi.BMI>22,4,3))\n", + "data_list.append(dd_bmi[['eid','assessment_date','bmi_status']].astype(int).to_numpy())\n", + "\n", + "dd_sm = dd[['assessment_date','smoking']].dropna().reset_index()\n", + "dd_sm = dd_sm[dd_sm['smoking']!=-3]\n", + "dd_sm['smoking_status'] = np.where(dd_sm['smoking']==1,8,np.where(dd_sm.smoking==2,7,6))\n", + "data_list.append(dd_sm[['eid','assessment_date','smoking_status']].astype(int).to_numpy())\n", + "\n", + "dd_al = dd[['assessment_date','alcohol']].dropna().reset_index()\n", + "dd_al = dd_al[dd_al['alcohol']!=-3]\n", + "dd_al['alcohol_status'] = np.where(dd_al['alcohol']==1,11,np.where(dd_al.alcohol < 4,10,9))\n", + "data_list.append(dd_al[['eid','assessment_date','alcohol_status']].astype(int).to_numpy())" ] }, { - "cell_type": "code", - "execution_count": null, - "id": "edeb7116-d937-4e8a-90dd-fff159e64a07", + "cell_type": "markdown", + "id": "7a156579-7132-4f26-899f-7ef0506894bf", "metadata": {}, + "source": [ + "## reformat, split train and val and output to delphi format" + ] + }, + { + "cell_type": "code", + "execution_count": 13, + "id": "69fa6d58-2861-4e5e-a480-2d6cb366ca86", + "metadata": { + "tags": [] + }, "outputs": [], "source": [ - "# Deal with all first occurances - this takes a long time\n", - "for i in tqdm(range(0,len(fo_fields))):\n", - " f = fo_fields[i]\n", - " d = df.select(['eid', f.name]).where(df[f.name].isNotNull())\n", - " d1 = d.withColumn(\"token\", lit(f.name))\n", - " d1 = d1.withColumnRenamed(f.name, \"date\")\n", - " d1 = d1.withColumn(\"age\", age_event_udf(col(\"eid\"), col(\"date\")))\n", - " d_all = d_all.union(d1)\n" + "data= np.vstack(data_list)\n", + "data = data[np.lexsort((data[:,1], data[:,2]==data[:,2].max(), data[:,0]))]\n", + "data = data[data[:,1]>=0]\n", + "data = pd.DataFrame(data).drop_duplicates([0,2]).values\n", + "data = data.astype(np.uint32)\n", + "data.tofile(output_prefix + '.bin')\n", + "ids = list(set(data[:,0]))\n", + "ids.sort()\n", + "train_val_split = data[:,0] <= ids[int(len(ids)*train_proportion)]\n", + "data[train_val_split].tofile(output_prefix + '_train.bin')\n", + "data[~train_val_split].tofile(output_prefix + '_val.bin')\n" ] }, { "cell_type": "code", "execution_count": null, - "id": "0770c44a-7585-42b3-8f48-cadba006c54f", + "id": "713a0abd-e5e3-40c3-b8b9-1921419f5eb3", "metadata": {}, "outputs": [], - "source": [ - "# Format, sort and write out to a file\n", - "df_all = d_all.select(\"eid\", \"age\", \"token\").toPandas()\n", - "df_all['age'] = df_all['age'] * 365.25\n", - "data = np.array(df_all).squeeze()\n", - "data[:,0] = data[:,0].astype(np.uint32)\n", - "data[:,1] = data[:,1].astype(np.uint32)\n", - "data = data[np.lexsort((data[:,1], data[:,0]))]\n", - "\n", - "data.tofile('all_records.bin')\n" - ] + "source": [] } ], "metadata": {