2020年10月20日 在エンジニアリングのブログ
低コストのゲノムシークエンスやAIを活用した医療用画像診断の普及により,精密医療への関心が高まっています。砖では、精密医療の領域において、データや AI を活用して疾患に対する最適な治療法を発見することを目指しています。精密医療は、希少疾患やがんと診断された患者の治療のアウトカムを改善してきましたが、精密医療はリアクティブ型の医療です。精密医療を受けるには、患者が病気である必要があります。
医療·ヘルスケアのコストとアウトカムの面では，糖尿病や心臓病，薬物使用障害などの慢性疾患の予防が，医療費と生活の質の改善に大きく影響を与えることがわかっています。米国では，10人死亡者ののうち7人が慢性疾患の患者で,医療費の85%が慢性疾患の治療によるものです。また，欧州や東南アジアでも同様の傾向が見られます。非感染性疾患は,患者への教育や慢性疾患の原因となる根本的な問題に対処することで,通常は予防可能です。これらの問題には，神経疾患の原因となる既知の遺伝的リスクなどの生物学的リスク，環境汚染や健康的な食事，予防治療へのアクセスの欠如などの社会経済的要因，喫煙や飲酒，座りっぱなしのラaaplフスタaaplルなどの行動によるリスクが含まれます。
精密医療は,データを利用して疾病発症のリスクが高い患者集団を特定し,疾病リスクを軽減する介入を行うことに重点を置いています。介入の種類には，リスクのある患者をリモトで監視してラフスタルや治療法を提案するデジタルアプリや,病状の監視の強化、補助的な予防医療の提供などがあります。しかし，これらの介入を実現するには，まずはリスクのある患者を特定する必要があります。
ハリスク患者を特定する優れたルの1に，タ(rwd)の活用があります。RWDとは，ヘルスケアのエコシステムで生成されるデ，タを相称したものです。電子医療記録(EMR)や健康記録(EHR)など,入院,臨床現場,薬局,医療従事者からのデータに加え,ゲノミクス,ソーシャルメディア,ウェアラブルなどの他のソースから収集されるデータも増えています。同じシリ，ズのこらのブログでは，ehrデタから臨床デタレクを構築する方法をご紹介しました。このブログ記事では,砖の統合データ分析プラットフォームを活用して臨床データレイクを構築して患者のジャーニーを追跡し,機械学習モデルを作成します。このモデルを使用して,患者の来歴やデモグラフィック情報を考慮すると,特定の期間内に特定の症状を発症した患者のリスクを評価できます。ここでは，薬物の過剰使用を例に取り上げます。物質使用障害によって引き起こされる広範な健康被害に関する重要なトピックです。MLflowを利用してモデルを追跡することで，モデルの経時的な変化を容易に把握できるため，モデルを患者ケアに展開するプロセスに信頼性がもたらされます。
特定の時間におけるリスクを予測するモデルをトレーニングするには,患者の関連するデモグラフィック情報(診断時の年齢,人種など),患者の診断履歴に関する時系列データを取得するデータセットが必要です。その後,このデータを利用して,患者が将来のある期間において病気と診断される可能性に影響を与える診断や,デモグラフィック的なリスクを学習するモデルをトレーニングできます。
図1:ehrから抽出されたデ，タスキ，マとテ，ブル間のリレ，ションシップ
このモデルをトレーニングするには,電子カルテ(EHR)に含まれるような患者の診断履歴やデモグラフィック情報を活用できます。図1は，今回のワクフロで使用するテブルを示しています。これらのテ，ブルは，前回のブログの笔记本を利用して作成したものです。三角洲湖から来歴、組織、患者データ（難読化された PII 情報を含む）をロードし、全患者のデモグラフィック情報を伴う来歴のデータフレームを作成します。
patient_encounters＝(遭遇。加入(病人,“病人”])。加入(组织、“组织”)显示(patient_encounters.filter(' reasondescription不是null ') .limit (10）)対象となる症状に基づいて，トレ，ニングデ，タに含めるのに適した患者のセットを選択します。来歴から,少なくとも1度は病気と診断された患者である症例と,病気の既往がない患者である対照を同じ数用意しています。
positive_patients＝(patient_encounters。选择（“病人”)。在哪里（较低的(“REASONDESCRIPTION”)。就像(“%{}%”.format (条件))) . dropduplicate () .withColumn(“is_positive”点燃(真正的))消极的病人＝(all_patients。加入(positive_patients在＝［“病人”),如何＝“left_anti”) .limit (positive_patients。数()) .withColumn (“is_positive”点燃(假()＝positive_patients.union (negative_patients)ここで，患者エンカウンタ，のセットを分析対象の患者に限定します。
qualified_patient_encounters_df＝(patient_encounters。加入(patients_to_study在＝［“病人”])。过滤器("DESCRIPTION is not NUll"))対象となるレコ，ドが揃ったので，次のステップでは特徴量を追加していきます。この予測タスクでは,デモグラフィック情報に加えて,ある症状や共存する症状(併存疾患)と診断された回数の合計,過去の訪問回数の数を選択します。
ほとんどの疾患では,併存疾患に関する多くの文献がありますが,RWDのデータセットを活用して対象疾患に関連する併存疾患を特定することも可能です。
comorbid_conditions＝(positive_patients。加入（patient_encounters，［“病人”])。在哪里(坳(“REASONDESCRIPTION”) .isNotNull ()) .dropDuplicates ([“病人”，“REASONDESCRIPTION”]) .groupBy (“REASONDESCRIPTION”）.数() .orderBy (“数”,提升＝假) .limit(num_conditions)私たの笔记本ウィジェットを使用して，含めるべき併存疾患の数や診察を行う時間の長さ(日単位)を指定しています。これらのパラメ，タはMLflowのトラッキングAPIを使って記録されます。
次に，それぞれの診察に特徴量を追加する必要があります。各併存疾患に対応して，対象の症状が過去に何回診察されたかを示す列を追加します。
ここでは
これらの特徴量を2のステップで追加します。まずは，併存疾患ンジゲタ関数を追加する関数(X我,c)を定義します。
Def add_comorbidities(qualified_patient_enccounters_df,comorbidity_list): output_df＝qualified_patient_encounters_df idx＝0为伴随疾病在comorbidity_list: output_df＝(output_df .withColumn("comorbidity_%d"％idx (output_df [“REASONDESCRIPTION”]．就像（“%”+伴随疾病(“REASONDESCRIPTION”］+“%”))。投（“int”) .withColumn(“comorbidity_ % d”％idx,合并(坳(“comorbidity_ % d”％idx),点燃(0)) #替换零值与0.cache()) idx+＝1返回(output_df)次に，Spark SQLの強力なウィンドウ関数のサポ，トを利用して，連続する日数の範囲でこれらの指標関数を合計します。
Def add_recent_encounters(encounter_features): lowest_date＝(encounter_features。选择（“START_TIME”) .orderBy (“START_TIME”) .limit (1) .withColumnRenamed (“START_TIME”，“EARLIEST_TIME”)) output_df＝(meet_features .crossJoin(lowest_date) .withColumn("day"， datediff(col(“START_TIME”)、坳(“EARLIEST_TIME”)) .withColumn("patient_age"， datediff(col(“START_TIME”)、坳(“生日”)))) w＝(Window.orderBy (output_df [“天”]) .partitionBy (output_df [“病人”]) .rangeBetween (-int(num_days),－1） ）为comorbidity_idx在范围(num_conditions): col_name＝“recent_ % d”％comorbidity_idx output_df＝(output_df .withColumn(col_name，总和(坳(“comorbidity_ % d”％comorbidity_idx))。在(w)) .withColumn (col_name,合并(坳(col_name),点燃(0))))返回(output_df)併存疾患の特徴量を追加した後は，タ，ゲット変数を追加する必要があります。ターゲット変数は,患者が将来の所定の時間枠内(例えば,現在の診察から1か月後)にターゲット疾患と診断されるかどうかを示します。このオペレーションのロジックは,前のステップと類似していますが,時間枠が未来のイベントをカバーしている点が異なります。対象とする症状が将来診断されるかどうかを示す，バ。
Def add_label(meet_features,num_days_future): w＝(Window.orderBy (encounter_features [“天”]) .partitionBy (encounter_features [“病人”]) .rangeBetween (0，num_days_future))＝(meet_features .withColumn(“标签”，马克斯(坳(“comorbidity_0”))。在(w)) .withColumn (“标签”，合并(坳(“标签”),点燃(0))))返回(output_df)次に，これらの特徴量を三角洲湖の特徴量ストアに書き込みます。再現性を確保するために，特徴量ストアのカラムとして，MLFlowの実験IDとランIDを追加します。このアプローチによるメリットは,より多くのデータを取得することで,将来参照するために再利用できる新たな特徴量を特徴量ストアに追加できることです。
トレ，ニングタスクに進む前に，クラス間で異なるラベルがどのように分布されているかを確認します。バイナリ分類の多くのアプリケーションでは,病気の予測などにおいて一方のクラスがレアになることがあります。このクラスの不均衡が，学習プロセスにマ。推定プロセスにおいて,モデルは多数派のクラスに焦点を合わせてレアなイベントを犠さらに，評価プロセスも損なわれます。例えば0/1ラベルが95%と5%に分布している不均衡のデータセットでは,モデルは常に0を予測し,精度は95%です。ラベルに不均衡がある場合，不均衡なデ，タを修正するために一般的なテクニックを適用する必要があります。
トレ，ニングデ，タを見てみると，デ，タセットが極めて不均衡であることがわかります。タesc escムウィンドウの95%以上が診断の根拠を示していません。この不均衡を調整するために，コントロ，ルクラスのダウンサンプルを行うか，合成サンプルを生成します。この選択はデ，タセットの規模や特徴量の数に依存します。この例では，多数派のダウンサンプルを行い，バランスのとれたデ，タセットを取得します。実際には，多数派のクラスのダウンサンプリングし，トレ，ニングアルゴリズムでクラスに重みを割り当てるなどメソッドを組み合わせる選択肢もあります。
df1＝dataset_df.filter (“标签= = 1”) n_df1＝df1。数() df2＝dataset_df.filter (“标签= = 0”)采样(假，0.9) .limit training_dataset_df (n_df1)＝df1.union (df2)采样(假，1．0)显示(training_dataset_df.groupBy (“标签”）.数（）)
モデルをトレーニングするには,デモグラフィックと併存疾患の特徴量のサブセットで条件を補強してそれぞれの観測値にラベルを適用し,このデータをモデルに渡してダウンストリームのトレーニングを行います。例えば,ここでは,最近診断された併存疾患に加えて,エンカウンタークラス(例:予約の理由,予防治療/ ER受診)および受診費用を追加し,デモグラフィック情報として,人種,性別,郵便番号,受診時の患者の年齢を選択しています。
多くの場合,元の臨床デSparkのデ，タフレ，ムを熊猫のデ，タフレ，ムに容易に変換し，任意のアルゴリズムに基づいてモデルをトレ，ニングできます。数据库の機械学習ランタ电子词典ムを利用する場合は，広範なオ，プンmlラ，ブラリを容易に利用できます。
あらゆる機械学習アルゴリズムがパラメータのセット(ハイパーパラメータ)を受け取り,入力パラメータに応じてスコアが変動します。また，パラメ，タやアルゴリズムを間違えると，オ，バ，フィッティングの問題を起こすケ，スもあります。モデルの性能を確保するために，ここではハパパラメタチュニングを使用し，最適なモデルア，キテクチャを選択します。その後，このステップで取得したパラメ，タ，を指定して最終的なモデルトレ，ニングを行います。
モデルチュ，ニングを行うには，まずデ，タの前処理が必要です。このデータセットには,数値の特徴量(例えば,最近の併存疾患の数)の他に,カテゴリ変数のデモグラフィックデータがあります。カテゴリ変数に最適なアプロ，チは，一热エンコディングの使用です。これには2理由があります。第一は，ほとんどの分類器(この場合はロジスティック回帰)は数値の特徴量を取り扱うからです。第二は,カテゴリー変数を単純に数値指標に変換すると,データに序列性が生じ,分類器を誤認させる可能性があるからです。例えば,カリフォルニア州を5ニューヨーク州を23のように,州名を指標に変換すると,ニューヨーク州はカリフォルニア州よりも”大きい”ことになります。アルファベット順にリスト化した州名の指標を反映していますが,このモデルのコンテキストでは何の意味もありません。一热エンコ，ディングでは，このような効果を排除できます。
この場合の前処理のステップは，入力パラメ，タを必要としません。ハパパラメタは分類器のみに影響を与えるため，前処理には影響しません。前処理は別で実行し，モデルのチュ，ニングには，その結果得たデ，タセットを使用します。
从sklearn.preprocessing进口OneHotEncoder进口numpy作为npdefpre_process（training_dataset_pdf)：X_pdf = training_dataset_pdf.drop (“标签”轴=1) y_pdf = training_dataset_pdf [“标签”= onehotencoder (handle_unknown=“忽略”X=one_hot_model.transform(X_pdf) y=y_pdf.values返回(X, y)次に，モデルに最適なパラメ，タを選択します。この分類には，ロジスティック回帰を弾性ネット正則化と一緒に使用します。在一个炎热エンコーディングを適用した後,問題のカテゴリ変数のカーディナリティに応じて,サンプル数を超える多くの特徴量を取得する可能性があります。このような問題に対するオ，バ，フィッティングを避けるには目的関数にペナルティを適用します。弾性ネット正則化のメリットは，2のペナルティ化手法(套索とリッジ回帰)を組み合わせることと,パイパーパラメータチューニングの過程で,混合の度合いを単一の変数で制御できることです。
このモデルを改善するために，hyperoptを使用してハパパラメタのグリッドを検索し，最適なパラメタを見けます。さらに，hyperoptのSparkTrialsモドを使用して，ハパパラメタを並列に検索します。このプロセスは,砖のマネージドMLflowを利用して,それぞれのハイパーパラメータの施行のおけるパラメータとメトリクスを自動的に記録します。各セットのパラメ，タのを検証するためには，モデルの評価の指標としてk-分割交差検証をF1スコアを用いて実行します。k——分割交差検証は複数の値が生成されるため,hyperoptを使用する際はスコアの最小値(ワーストケースのシナリオ)を選択し,それを最大化するようにしていることにご注意ください。
从数学导入exp def params_to_lr(params):返回｛“惩罚”：“elasticnet”，“multi_class”：“表达”，“random_state”：43，“n_jobs”：－1，“规划求解”：“传奇”，“托尔”：经验值(params [“托尔”]), #经验值()这里是因为超参数是在日志空间“C”：经验值(params [“C”]),“l1_ratio”：经验值(params [“l1_ratio”} def tune_model(params):与mlflow.start_run (run_name＝“tunning-logistic-regression”、嵌套＝真正的）作为运行:clf＝LogisticRegression (＊＊params_to_lr (params))。拟合(X, y)损失＝-cross_val_score(clf, X, y,n_jobs＝－1,得分＝“f1”）.最小值（）返回｛“状态”: STATUS_OK,“损失”:损失}探索空間の検索を改善するために,logspaceのパラメータのグリッドを選択し,hyperoptにより提案されたパラメータを変換するための変換関数を定義します。アプロチの全体の概要と，ハパパラメタ空間の定義をこのように選択した理由にいては，こらのト，クでご覧いただけます。このトークでは,砖上でエンドツーエンドの機械学習ライフサイクルを管理する方法を解説しています。
从导入fmin, hp, tpe, SparkTrials, STATUS_OK search_space＝{#使用制服在这里是Loguniform来使指标显示了更好的在mlflow比较,在logspace“托尔”: hp.uniform (“托尔”，3，0)，“C”: hp.uniform (“C”，－2，0)，“l1_ratio”: hp.uniform (“l1_ratio”，3，－1)，} spark_trials＝SparkTrials(并行性＝2) best_params＝fmin (fn＝tune_model、空间＝search_space,算法＝山丘。建议,max_evals＝32, rstate＝np.random.RandomState (43),试验＝spark_trials)この実行で得る成果は，交差検証によるf1スコアに基づいて評価された最適なパラメ，タです。
params_to_lr (best_params)出［46]: {“惩罚”：“elasticnet”，“multi_class”：“表达”，“random_state”：43，“n_jobs”：－1，“规划求解”：“传奇”，“托尔”：0.06555920596441883，“C”：0.17868321158011416，“l1_ratio”：0.27598949120226646｝次に，MLflowのダッシュボ，ドを見てみましょう。MLflowは hyperopt の実行を自動的にグループ化します。そのため、図3のようにさまざまなプロットを用いて、各ハイパーパラメータが損失関数に与える影響を調べることができます。この機能は、モデルの動向の理解やハイパーパラメータによる効果を深く理解するために特に重要です。例えば、正則化の強さの逆数である C の値が低いと、F1 の値が高くなることに気づくことができます。
図3:MLflowで作成したモデルの平行座標
最適なパラメータの組み合わせを見つけた後は,最適なハイパーパラメータを用いてバイナリ分類器をトレーニングし,MLFLowでモデルをロギングします。MLflowのモデルAPIは,トレーニングに使用されたライブラリに関係なく,python関数として容易にモデルを格納し,後にモデルのスコアリングの際に呼び出すことができます。モデルの発見性を高めるために,対象となる条件に関連した名前を付けてモデルを記録しています。(ここでは,“薬物の過剰摂取”)
进口mlflow.sklearn进口matplotlib.pyplot作为plt从sklearn.pipeline进口管道从mlflow.models.signature进口因为我们希望模型输出概率(风险)而不是预测的标签，所以我们覆盖了## mlflow.pyfun类SklearnModelWrapper(mlflow. pyfuncc . pythonmodel): def __init__(self, model): self。Model = Model def predict(self, context, model_input): return self. Model .predict_proba(model_input)[:，1] def train(params): with mlflow.start_run(run_name='training-logistic-regression'，嵌套=True)作为运行:mlflow.log_params(params_to_lr(params)) X_arr=training_dataset_pdf.drop('标签”,轴= 1)。值y_arr=training_dataset_pdf['标签']。values ohe = OneHotEncoder(handle_unknown='忽略') clf = LogisticRegression(**params_to_lr(params))。fit(X, y) pipe = Pipeline([('在一个炎热的'， ohe)， ('clf'， clf)]) lr_model = pipe。cross (clf, ohe.transform(X_arr)， y_arr,n_jobs=-1, scores ='精度').mean() wrapped_lr_model = SklearnModelWrapper(lr_model) model_name= '-. join (condition.split ()) mlflow.log_metric ('精度mlflow. pyfunc_log_model (model_name, python_model=wrapped_lr_model) displayHTML('模型精度为: %s %(分数))返回(mlflow.active_run () . info)ここで，前のステップで得られた最適なパラメ，タを指定して，モデルをトレ，ニングします。
モデルのトレーニングでは,sklearnのパイプラインの一部として,前処理(一个炎热エンコーディング)を含め,エンコーダーと分類器を1つのモデルとして記録しています。次のステップでは，シンプルに患者デ，タのモデルを呼び出して，患者のリスクを評価します。
モデルをトレーニングしてMLflowに記録したら,次のステップは,そのモデルを用いた新たなデータのスコアリングです。MLflowでは，異なるタグを使用して実験を検索できる機能があります。例えばこの場合,モデルのトレーニングに指定された実行名を使用して,トレーニングモデルのアーティフその後，主要な指標に基づいて取得した実験を並べ替えることができます。
进口mlflow best_run = mlflow.search_runs (filter_string ="tags.mlflow.runName = 'training- logistics -regression'"order_by = (的指标。精度DESC']) .iloc [0] model_name =药物过量的clf = mlflow.pyfunc.load_model (model_uri =“% s / % s”% (best_run.artifact_uri model_name) clf_udf = mlflow.pyfunc。spark_udf(火花,model_uri =“% s / % s”% (best_run.artifact_uri model_name))特定のモデルを選択したら，モデルのuriや名前を指定してモデルをロ，ドします。
砖のモデルレジストリを利用して,モデルのバー
このブログ記事では最初に,慢性疾患の発症を促進する臨床的およびデモグラフィック的な共変量を特定する精密予防システムの必要性について説明しました。次に，EHR から生成されたシミュレーションデータを用いて、薬物過剰摂取のリスクがある患者を特定するエンドツーエンドの機械学習ワークフローを見てきました。このワークフローの最後では、MLflowから学習した機械学習モデルをエクスポートすることに成功し、患者データの新たなストリームに適用しました。
このモデルは参考にはなりますが，実際のプラクティスに移行できなければ成果はありません。リアルワ，ルドの実践では，多くのお客様と協業し，これらと同様のシステムを実運用に展開してきました。例えば,サウスカロライナ医科大学では,EHRデータを処理して敗血症のリスクがある患者を特定するライブストリーミングパイプラインを展開しています。これにより，敗血症による患者の衰弱を8時間前に発見できるようになりました。INTEGRIS健康社では,同様のシステムを活用して,褥瘡(床ずれ)発生の兆候をEHRデータでモニタリングしています。どらの環境でも，患者が特定されると，ケアチムにその症状がすぐに通知されます。医療保険の領域では，Optum社と協業して同様のモデルを展開しています。社为联合健康集团では,長期,短期のアーキテクチャの回帰型ニューラルネットワークを使用する疾患予測エンジンを開発し,9つの異なる疾患領域にわたって一般化を伴う疾患の進行を同定していこのモデルは,患者に予防ケアのパスを割り当てるために用いられ,慢性疾患患者のアウトカムと治療コストの改善につながりました。
ここまでは,ヘルスケア領域における疾病予測アルゴリズムの活用にフォーカスしていましたが,これらのモデルを製薬業界のために構築して展開する機会も大いにあります。疾病予測モデルは，市販後に医薬品がどのように使用されているかの分析情報を提供します。また，これまで検知されなかった保護効果を検知し，ラベル表記の拡大に役立てることができます。さらに，疾病予測モデルを活用して，希少疾患や診断が不十分な疾患の臨床試験への登録を検討することも可能です。希少疾患の診断を受ける前に誤診された患者を検知するモデルを構築することで,臨床医に一般的な誤診のパターンを教育する教材を作成できます。また,試験の登録者数と有効性の改善につながる試験の包括的な基準を作成できるようになることも期待されています。
このブログ記事では,慢性疾患の発症リスクを抱える患者を特定するために,リアルワールドのデータに機械学習を活用する方法を解説しました。医療デ，タセットを格納，処理するための三角洲湖に関しては，リアルワ，ルド医療デ，タで詳細を説明しています。また，本ブログで使用した患者リスクのスコアリング笔记本をダウンロドして，無料トラアルを開始できます。
