ニューラルネットワークは、オブジェクトの検出、画像の類似性、分類などの複雑なコンピュータビジョンタスクの解決に効果的であることが証明されています。 低コストのGPUの進化に伴い、ニューラルネットワークの構築と展開の計算コストが大幅に削減されました。 ただし、ほとんどの手法は、ビジュアルメディアで一般的に見られるピクセル解像度を処理するように設計されています。 たとえば、一般的な解像度サイズは、YOLOv544の場合は416ピクセルと3ピクセル、SSDの場合は300ピクセルと512ピクセル、VGGの場合は224ピクセルです。 衛星画像やデジタル病理画像などのギガピクセル画像（10 ^ 9 +ピクセル）で構成されるデータセットに対して分類器をトレーニングすることは、計算上困難です。 各GPUは使用可能なメモリによって制限されるため、これらの画像をニューラルネットワークに直接入力することはできません。 これには、元の画像を小さなチャンクで処理できるようにするために、タイリングなどの特定の前処理技術が必要です。 さらに、これらの画像のサイズが大きいため、全体的なトレーニング時間は長くなる傾向があり、分散トレーニングなどの適切なスケーリング手法を使用しないと、数日または数週間かかることがよくあります。

この投稿では、次のXNUMXつの目的を達成するために、拡張性の高い機械学習（ML）パイプラインを構築する方法について説明します。

データセット

この投稿では、 データセット から取得した全スライドデジタル病理画像で構成されています がんゲノムアトラス （TCGA）LUAD（腺癌）、LUSC（扁平上皮癌）、または正常な肺組織として正確かつ自動的に分類します。ここで、LUADとLUSCは肺癌のXNUMXつの最も一般的なサブタイプです。 データセットは、NIHおよびNCIによって公開されています。

生の高解像度画像は SVS フォーマット。 SVSファイルは、Aperio顕微鏡画像のアーカイブと分析に使用されます。 この投稿で使用されている手法とツールは、衛星画像を含むあらゆる超高解像度画像データセットに適用できます。

以下は、組織スライドのサンプル画像です。 この単一の画像には750億ピクセル以上が含まれ、XNUMXMB以上のメモリを占有します。 この画像は元の形式でニューラルネットワークに直接供給することができないため、画像を多数の小さな画像に並べて表示する必要があります。

以下は、前の組織スライド画像を前処理した後に生成されたタイル画像のサンプルです。 これらのRGB3チャンネル画像のサイズは512×512で、ニューラルネットワークへの入力として直接使用できます。 これらのタイル画像にはそれぞれ、親スライドと同じラベルが割り当てられています。 さらに、背景が50％を超えるタイル画像は破棄されます。

アーキテクチャの概要

次の図は、元の生の画像から推論まで、全体的なエンドツーエンドのアーキテクチャを示しています。 まず、SageMaker Processingを使用して、画像を並べて表示し、ズームして、トレーニングとテストスプリットに並べ替えてから、分散SageMakerトレーニングに必要な数のシャードにパッケージ化します。 次に、SageMakerトレーニングジョブはDockerコンテナをからロードします Amazon エラスティック コンテナ レジストリ （Amazon ECR）。 このジョブは、パイプモードを使用して、準備された画像の断片からデータを読み取り、モデルをトレーニングし、最終的なモデルアーティファクトをに保存します。 Amazon シンプル ストレージ サービス （Amazon S3）。 最後に、トレーニング済みのモデルをリアルタイムの推論エンドポイントにデプロイします。このエンドポイントは、適切なDockerコンテナ（Amazon ECRから）とモデル（Amazon S3から）をロードして、低レイテンシで推論リクエストを処理します。

SageMaker処理を使用したデータ前処理

SVSスライド画像は次のXNUMXつのステップで前処理されます。

• 画像のタイリング –画像は重複しない512×512ピクセルのウィンドウで並べて表示され、背景が50％を超えるタイルは破棄されます。 タイルはJPEG画像として保存されます。
• 画像をTFRecordsに変換する – SageMakerパイプモードを使用してトレーニング時間を短縮します。これには、データがプロトバッファー形式で利用可能である必要があります。 TFRecordは、TensorFlowを使用してモデルをトレーニングするために使用される一般的なプロトバッファー形式です。 次のセクションでは、SageMakerパイプモードとプロトバッファ形式について詳しく説明します。
• TFRecordの並べ替え –データセットを、XNUMX方向分類器（LUAD / LUSC / Normal）のテスト、トレーニング、および検証のコホートに分類します。 TCGAデータセットには、XNUMX人の患者に対応する複数のスライド画像を含めることができます。 データ漏えいを防ぐために、同じ患者に対応するスライドから生成されたすべてのタイルが同じ分割を占めるようにする必要があります。 テストセットでは、そのスライドのすべてのタイルを含むスライドごとのTFRecordを作成して、展開で使用される方法でモデルを評価できるようにします。

以下は前処理コードです。

def generate_tf_records(base_folder, input_files, output_file, n_image, slide=None): record_file = output_file count = n_image with tf.io.TFRecordWriter(record_file) as writer: while count: filename, label = random.choice(input_files) temp_img = plt.imread(os.path.join(base_folder, filename)) if temp_img.shape != (512, 512, 3): continue count -= 1 image_string = np.float32(temp_img).tobytes() slide_string = slide.encode('utf-8') if slide else None tf_example = image_example(image_string, label, slide_string) writer.write(tf_example.SerializeToString())

前の前処理ステップにはSageMakerProcessingを使用します。これにより、SageMakerを使用して、データの前処理または後処理、機能エンジニアリング、データ検証、モデル評価のワークロードを実行できます。 処理ジョブは、Amazon S3からのデータを入力として受け入れ、処理された出力データをAmazonS3に保存します。

SageMaker Processingを使用する利点は、複数のコンピューティングインスタンスに入力を簡単に分散できることです。 簡単に設定できます s3_data_distribution_type=ShardedByS3Key すべての処理コンテナ間でデータを均等に分割するパラメータ。

重要なのは、処理インスタンスの数が、Horovodを使用した分散トレーニングに使用するGPUの数（つまり、16）と一致することです。 Horovodトレーニングを導入すると、その理由がより明確になります。

処理スクリプトはで利用可能です GitHubの.

processor = Processor(image_uri=image_name, role=get_execution_role(), instance_count=16, # run the job on 16 instances base_job_name='processing-base', # should be unique name instance_type='ml.m5.4xlarge', volume_size_in_gb=1000) processor.run(inputs=[ProcessingInput( source=f's3://<bucket_name>/tcga-svs', # s3 input prefix s3_data_type='S3Prefix', s3_input_mode='File', s3_data_distribution_type='ShardedByS3Key', # Split the data across instances destination='/opt/ml/processing/input')], # local path on the container outputs=[ProcessingOutput( source='/opt/ml/processing/output', # local output path on the container destination=f's3://<bucket_name>/tcga-svs-tfrecords/' # output s3 location )], arguments=['10000'], # number of tiled images per TF record for training dataset wait=True, logs=True)

SageMakerトレーニングを使用した分散モデルトレーニング

MLモデルを概念化から本番環境に移行することは、通常、複雑で時間がかかります。 モデルをトレーニングするには大量のデータを管理し、トレーニングに最適なアルゴリズムを選択し、トレーニング中に計算能力を管理してから、モデルを本番環境にデプロイする必要があります。 SageMakerは、次のことをはるかに簡単にすることで、この複雑さを軽減します MLモデルを構築してデプロイする。 基盤となるインフラストラクチャを管理して、モデルをペタバイト規模でトレーニングし、本番環境にデプロイします。

スライド全体の画像を前処理した後でも、数百ギガバイトのデータが残っています。 単一インスタンス（GPUまたはCPU）でのトレーニングは、完了するまでに数日または数週間かかります。 処理を高速化するには、モデルのトレーニングのワークロードを複数のインスタンスに分散させる必要があります。 この投稿では、分散トレーニングフレームワークであるHorovodとSageMakerパイプモードを使用したデータ並列処理に基づく分散ディープラーニングに焦点を当てます。

Horovod：クロスプラットフォームの分散トレーニングフレームワーク

大量のデータを使用してモデルをトレーニングする場合、データは単一インスタンスまたは複数インスタンスのいずれかで複数のCPUまたはGPUに分散する必要があります。 ディープラーニングフレームワークは、分散トレーニングをサポートする独自の方法を提供します。 ホロヴォード は、分散型深層学習のための人気のあるフレームワークにとらわれないツールキットです。 これは、（パラメーターサーバーアプローチと比較して）高速分散トレーニングにallreduceアルゴリズムを利用し、分散トレーニングを高速化するための複数の最適化手法を備えています。 SageMakerでのHorovodを使用した分散トレーニングのその他の例については、を参照してください。 Amazon SageMakerパイプモードでHorovodを使用したマルチGPUおよび分散トレーニング & Amazon SageMakerでApache MXNetとHorovodを使用してトレーニング時間を短縮する.

SageMakerパイプモード

ファイルモードまたはパイプモードのいずれかでSageMakerに入力を提供できます。 ファイルモードでは、入力ファイルがトレーニングインスタンスにコピーされます。 と パイプモード、データセットはトレーニングインスタンスに直接ストリーミングされます。 これは、トレーニングジョブがより早く開始され、計算とダウンロードが並行して実行され、必要なディスク容量が少なくなることを意味します。 したがって、大規模なデータセットにはパイプモードをお勧めします。

SageMakerパイプモードでは、データがプロトコルバッファ形式である必要があります。 プロトコルバッファ 構造化データをシリアル化するための、言語に依存しない、プラットフォームに依存しない、拡張可能なメカニズムです。 TFレコード は、TensorFlowを使用してモデルをトレーニングするために使用される一般的なプロトバッファ形式です。 TFRecordは、TensorFlowで複数の方法で使用できるように最適化されています。 まず、複数のデータセットを簡単に組み合わせて、ライブラリが提供するデータのインポートおよび前処理機能とシームレスに統合できます。 次に、このタイプのデータの非常に効率的で（コーディングの観点から）便利なインポートを可能にする方法で、シーケンスデータ（時系列や単語エンコーディングなど）を格納できます。

次の図は、パイプモードでのデータアクセスを示しています。

SageMakerパイプモードでのデータシャーディング

SageMakerパイプモードとHorovodを使用する場合は、いくつかの考慮事項に留意する必要があります。

• 各パイプを介してストリーミングされるデータは、他のパイプと相互に排他的です。 パイプの数によって、作成する必要のあるデータシャードの数が決まります。
• Horovodは、各コンピューティングインスタンスのトレーニングスクリプトをラップします。 これは、各コンピューティングインスタンスのデータが異なるシャードからのものである必要があることを意味します。
• SageMakerトレーニングパラメーターを使用 S3DataDistributionType に設定 ShardedByS3Key、パイプを複数のインスタンスと共有できます。 データは、インスタンス間でラウンドロビン方式でストリーミングされます。

これをわかりやすく説明するために、ml.p3.8xlargeタイプの3.8つのインスタンス（AとB）を使用するとします。 各ml.p1xlargeインスタンスには2つのGPUがあります。 3本のパイプ（P4、PXNUMX、PXNUMX、PXNUMX）を作成し、 S3DataDistributionType = 'ShardedByS3Key’。 次の表に示すように、各パイプはラウンドロビン方式でXNUMXつのインスタンス間でデータを均等に分散します。 これは、Horovodでパイプを設定する際に必要なコアコンセプトです。 Horovodは各GPUのトレーニングスクリプトをラップするため、トレーニングインスタンスごとにGPUと同じ数のパイプを作成する必要があります。

次のコードは、各パイプのAmazonS3のデータをシャーディングします。 Amazon S3では、各シャードに個別のプレフィックスが必要です。

# Definite distributed training hyperparameters
train_instance_type='ml.p3.8xlarge'
train_instance_count = 4
gpus_per_host = 4
num_of_shards = gpus_per_host * train_instance_count distributions = {'mpi': { 'enabled': True, 'processes_per_host': gpus_per_host }
}

# Sharding
client = boto3.client('s3')
result = client.list_objects(Bucket=s3://<bucket_name>, Prefix='tcga-svs-tfrecords/train/', Delimiter='/') j = -1
for i in range(num_of_shards): copy_source = { 'Bucket': s3://<bucket_name>, 'Key': result['Contents'][i]['Key'] } print(result['Contents'][i]['Key']) if i % gpus_per_host == 0: j += 1 dest = 'tcga-svs-tfrecords/train_sharded/' + str(j) +'/' + result['Contents'][i]['Key'].split('/')[2] print(dest) s3.meta.client.copy(copy_source, s3://<bucket_name>, dest) # Define inputs to SageMaker estimator
svs_tf_sharded = f's3://<bucket_name>/tcga-svs-tfrecords'
shuffle_config = sagemaker.session.ShuffleConfig(234)
train_s3_uri_prefix = svs_tf_sharded
remote_inputs = {} for idx in range(gpus_per_host): train_s3_uri = f'{train_s3_uri_prefix}/train_sharded/{idx}/' train_s3_input = s3_input(train_s3_uri, distribution ='ShardedByS3Key', shuffle_config=shuffle_config) remote_inputs[f'train_{idx}'] = train_s3_input remote_inputs['valid_{}'.format(idx)] = '{}/valid'.format(svs_tf_sharded)
remote_inputs['test'] = '{}/test'.format(svs_tf_sharded)
remote_inputs

SageMaker Estimatorを使用して、ml.p3.8xlargeの16つのインスタンスでトレーニングを開始します。 各インスタンスにはXNUMXつのGPUがあります。 したがって、合計XNUMX個のGPUがあります。 次のコードを参照してください。

local_hyperparameters = {'epochs': 5, 'batch-size' : 16, 'num-train':160000, 'num-val':8192, 'num-test':8192} estimator_dist = TensorFlow(base_job_name='svs-horovod-cloud-pipe', entry_point='src/train.py', role=role, framework_version='2.1.0', py_version='py3', distribution=distributions, volume_size=1024, hyperparameters=local_hyperparameters, output_path=f's3://<bucket_name>/output/', instance_count=4, instance_type=train_instance_type, input_mode='Pipe') estimator_dist.fit(remote_inputs, wait=True)

トレーニングスクリプトの次のコードスニペットは、分散トレーニングのためにTensorFlowを使用してHorovodをオーケストレーションする方法を示しています。

mpi = False
if 'sagemaker_mpi_enabled' in args.fw_params: if args.fw_params['sagemaker_mpi_enabled']: import horovod.keras as hvd mpi = True # Horovod: initialize Horovod. hvd.init() # Pin GPU to be used to process local rank (one GPU per process) gpus = tf.config.experimental.list_physical_devices('GPU') tf.config.experimental.set_visible_devices(gpus[hvd.local_rank()], 'GPU')
else: hvd = None callbacks = []
if mpi: callbacks.append(hvd.callbacks.BroadcastGlobalVariablesCallback(0)) callbacks.append(hvd.callbacks.MetricAverageCallback()) if hvd.rank() == 0: callbacks.append(ModelCheckpoint(args.output_dir + '/checkpoint-{epoch}.ckpt', save_weights_only=True, verbose=2))
else: callbacks.append(ModelCheckpoint(args.output_dir + '/checkpoint-{epoch}.ckpt', save_weights_only=True, verbose=2)) train_dataset = train_input_fn(hvd, mpi)
valid_dataset = valid_input_fn(hvd, mpi)
test_dataset = test_input_fn()
model = model_def(args.learning_rate, mpi, hvd)
logging.info("Starting training")
size = 1
if mpi: size = hvd.size() model.fit(train_dataset, steps_per_epoch=((args.num_train // args.batch_size) // size), epochs=args.epochs, validation_data=valid_dataset, validation_steps=((args.num_val // args.batch_size) // size), callbacks=callbacks, verbose=2)

パイプモードはデータを各インスタンスにストリーミングするため、トレーニングスクリプトはトレーニング中にデータサイズを計算できません（計算に必要です）。 Steps_per_epoch）。 したがって、パラメーターはTensorFlow推定器へのハイパーパラメーターとして手動で提供されます。 さらに、GPU間で均等に分割できるように、データポイントの数を指定する必要があります。 各GPUがトレーニングプロセスを完了するのにかかる時間はもはや同じではないため、不均等な分割はHorovodデッドロックにつながる可能性があります。 データポイントが均等に分割されるようにするために、トレーニング用のGPUの数と同じ数のインスタンスを前処理に使用します。 この例では、この数は16です。

推論と展開

SageMakerを使用してモデルをトレーニングした後、新しいイメージで推論するためにモデルをデプロイします。 一度にXNUMXつの予測を取得するように永続エンドポイントを設定するには、SageMakerホスティングサービスを使用します。 データセット全体の予測を取得するには、SageMakerバッチ変換を使用します。

この投稿では、トレーニング済みモデルをSageMakerエンドポイントとしてデプロイします。 次のコードは、モデルをm4インスタンスにデプロイし、TFRecordsからタイル画像データを読み取り、スライドレベルの予測を生成します。

# Generate predictor object from trained model
predictor = estimator_dist.deploy(initial_instance_count=1, instance_type='ml.m4.xlarge') # Tile-level prediction
raw_image_dataset = tf.data.TFRecordDataset(f'images/{local_file}') # read a TFrecord
parsed_image_dataset = raw_image_dataset.map(dataset_parser) # Parse TFrecord to JPEGs pred_scores_list = []
for i, element in enumerate(parsed_image_dataset): image = element[0].numpy() label = element[1].numpy() slide = element[2].numpy().decode() if i == 0: print(f"Making tile-level predictions for slide: {slide}...") print(f"Querying endpoint for a prediction for tile {i+1}...") pred_scores = predictor.predict(np.expand_dims(image, axis=0))['predictions'][0] pred_class = np.argmax(pred_scores) if i > 0 and i % 10 == 0: plt.figure() plt.title(f'Tile {i} prediction: {pred_class}') plt.imshow(image / 255) pred_scores_list.append(pred_scores)
print("Done.") # Slide-level prediction (average score over all tiles)
mean_pred_scores = np.mean(np.vstack(pred_scores_list), axis=0)
mean_pred_class = np.argmax(mean_pred_scores)
print(f"Slide-level prediction for {slide}:", mean_pred_class)

モデルは、個々のタイル画像でトレーニングされます。 推論中に、SageMakerエンドポイントは各タイルの分類スコアを提供します。 これらのスコアはすべてのタイルで平均化され、スライドレベルのスコアと予測が生成されます。 次の図は、このワークフローを示しています。

多数決方式も適切です。

スライド画像の大規模な新しいバッチで推論を実行するには、複数のインスタンスのAmazonS3のデータセットでオフライン予測のバッチ変換ジョブを実行できます。 処理されたTFRecordがAmazonS3から取得されたら、前の手順を複製して、新しい画像ごとにスライドレベルの分類を生成できます。

まとめ

この投稿では、SageMaker Processing、SageMaker Pipeモード、およびHorovodを使用する超高解像度画像用のスケーラブルな機械学習パイプラインを紹介しました。 パイプラインは、ギガピクセルスケールに近づく画像で構成されるデータセットに対する分類器の大規模なトレーニングの複雑なプロセスを簡素化します。 SageMakerとHorovodを使用して、入力を複数のコンピューティングインスタンスに分散することでプロセスを簡素化し、トレーニング時間を短縮しました。 また、タイルレベルの予測を集約してスライドレベルの推論を生成するための、シンプルで効果的な戦略も提供しました。

SageMakerの詳細については、を参照してください。 Amazon SageMakerを使用して機械学習モデルを構築、トレーニング、デプロイします。 パイプモードとHorovodが一緒に適用されるSageMakerで実行する完全な例については、 GitHubレポ.

参考文献

1. Nicolas Coudray、Paolo Santiago Ocampo、Theodore Sakellaropoulos、Navneet Narula、Matija Snuderl、DavidFenyö、Andre L. Moreira、Narges Razavian、Aristotelis Tsirigos 「深層学習を使用した非小細胞肺癌の組織病理学的画像からの分類と突然変異の予測」。 ネイチャーメディシン、2018; DOI：10.1038 / s41591-018-0177-5
2. https://github.com/ncoudray/DeepPATH/tree/master/DeepPATH_code
3. https://www.cancer.gov/about-nci/organization/ccg/research/structural-genomics/tcga

著者について

カランシンドワニ はAmazonMachine Learning Solutionsのデータサイエンティストであり、ディープラーニングモデルを構築して展開しています。 彼はコンピュータービジョンの分野を専門としています。 余暇にはハイキングを楽しんでいます。

ビナイ・ハヌマイア はAmazonML Solutions Labのディープラーニングアーキテクトであり、顧客がAIおよびMLソリューションを構築してビジネス上の課題を加速するのを支援しています。 これに先立ち、彼はAWSDeepLensとAmazonPersonalizeの立ち上げに貢献しました。 余暇には家族との時間を楽しんでおり、熱心なロッククライマーです。

ライアンブランド は、Amazon Machine Learning SolutionsLabのデータサイエンティストです。 彼は、ヘルスケアとライフサイエンスの問題に機械学習を適用する特定の経験があり、自由な時間には、歴史とサイエンスフィクションを読むことを楽しんでいます。

荒井達也博士 は、Amazon Machine Learning SolutionsLabチームの深層学習データサイエンティストに転向した生物医学エンジニアです。 彼は、AIの真の民主化を信じており、AIの力はコンピューター科学者や数学者だけに限定されるべきではないと信じています。

コインスマート。 BesteBitcoin-ヨーロッパのBörse
ソース：https：//aws.amazon.com/blogs/machine-learning/building-a-scalable-machine-learning-pipeline-for-ultra-high-resolution-medical-images-using-amazon-sagemaker/