Pelatihan yang Terdistribusi

Pelatihan terdistribusi adalah jenis pelatihan model di mana kebutuhan sumber daya komputasi (misalnya CPU, RAM) didistribusikan ke beberapa komputer. Pelatihan terdistribusi memungkinkan pelatihan lebih cepat dan pada kumpulan data yang lebih besar (hingga beberapa miliar contoh).

Pelatihan terdistribusi juga berguna untuk pengoptimalan hyper-parameter otomatis di mana beberapa model dilatih secara paralel.

Dalam dokumen ini Anda akan mempelajari cara:

  • Latih model TF-DF menggunakan pelatihan terdistribusi.
  • Sesuaikan hyper-parameter model TF-DF menggunakan pelatihan terdistribusi.

Keterbatasan

Saat ini, pelatihan terdistribusi didukung untuk:

  • Melatih model Gradient Boosted Trees dengan tfdf.keras.DistributedGradientBoostedTreesModel . Model Pohon Peningkatan Gradien Terdistribusi setara dengan model yang tidak terdistribusi.
  • Pencarian hyper-parameter untuk semua jenis model TF-DF.

Cara mengaktifkan pelatihan terdistribusi

Bagian ini mencantumkan langkah-langkah untuk mengaktifkan pelatihan terdistribusi. Untuk contoh selengkapnya, lihat bagian berikutnya.

Cakupan ParameterServerStrategy

Model dan himpunan data ditentukan dalam cakupan ParameterServerStrategy . 

strategy = tf.distribute.experimental.ParameterServerStrategy(...)
with strategy.scope():
  model = tfdf.keras.DistributedGradientBoostedTreesModel()
  distributed_train_dataset = strategy.distribute_datasets_from_function(dataset_fn)
model.fit(distributed_train_dataset)

Format kumpulan data

Seperti untuk pelatihan non-terdistribusi, kumpulan data dapat disediakan sebagai

  1. Kumpulan data terdistribusi tensorflow terbatas, atau
  2. jalur ke file kumpulan data menggunakan salah satu format kumpulan data yang kompatibel .

Menggunakan file shard jauh lebih sederhana dibandingkan menggunakan pendekatan kumpulan data terdistribusi tensorflow terbatas (1 baris vs ~20 baris kode). Namun, hanya pendekatan kumpulan data Tensorflow yang mendukung pra-pemrosesan TensorFlow. Jika pipeline Anda tidak berisi pra-pemrosesan apa pun, opsi sharded dataset direkomendasikan.

Dalam kedua kasus tersebut, kumpulan data harus dibagi menjadi beberapa file untuk mendistribusikan pembacaan kumpulan data secara efisien.

Pekerja pengaturan

Proses utamanya adalah program yang menjalankan kode python yang mendefinisikan model TensorFlow. Proses ini tidak menjalankan komputasi yang berat. Perhitungan pelatihan yang efektif dilakukan oleh pekerja . Pekerja adalah proses yang menjalankan Server Parameter TensorFlow.

Kepala harus dikonfigurasi dengan alamat IP pekerja. Hal ini dapat dilakukan dengan menggunakan variabel lingkungan TF_CONFIG , atau dengan membuat ClusterResolver . Lihat Pelatihan server parameter dengan ParameterServerStrategy untuk detail selengkapnya.

ParameterServerStrategy TensorFlow mendefinisikan dua jenis pekerja: "pekerja" dan "server parameter". TensorFlow memerlukan setidaknya satu dari setiap jenis pekerja untuk dibuat instance-nya. Namun, TF-DF hanya menggunakan "pekerja". Jadi, satu "server parameter" perlu dipakai tetapi tidak akan digunakan oleh TF-DF. Misalnya, konfigurasi pelatihan TF-DF mungkin terlihat seperti berikut:

  • 1 Ketua
  • 50 Pekerja
  • 1 Server parameter

Para pekerja memerlukan akses ke operasi pelatihan khusus TensorFlow Decision Forests. Ada dua opsi untuk mengaktifkan akses:

  1. Gunakan Server Parameter TF-DF C++ yang telah dikonfigurasi sebelumnya //third_party/tensorflow_decision_forests/tensorflow/distribute:tensorflow_std_server .
  2. Buat server parameter dengan memanggil tf.distribute.Server() . Dalam hal ini, TF-DF harus diimpor import tensorflow_decision_forests .

Contoh

Bagian ini menunjukkan contoh lengkap konfigurasi pelatihan terdistribusi. Untuk contoh lainnya, lihat pengujian unit TF-DF .

Contoh: Pelatihan terdistribusi pada jalur kumpulan data

Bagilah kumpulan data Anda menjadi sekumpulan file pecahan menggunakan salah satu format kumpulan data yang kompatibel . Disarankan untuk memberi nama file sebagai berikut: /path/to/dataset/train-<5 digit index>-of-<total files> , misalnya

/path/to/dataset/train-00000-of-00100
/path/to/dataset/train-00001-of-00005
/path/to/dataset/train-00002-of-00005
...

Untuk efisiensi maksimum, jumlah file harus minimal 10x jumlah pekerja. Misalnya, jika Anda berlatih dengan 100 pekerja, pastikan kumpulan data terbagi dalam setidaknya 1000 file.

File-file tersebut kemudian dapat direferensikan dengan ekspresi sharding seperti:

  • /path/ke/dataset/train@1000
  • /jalur/ke/kumpulan data/kereta@*

Pelatihan terdistribusi dilakukan sebagai berikut. Dalam contoh ini, kumpulan data disimpan sebagai TFRecord dari Contoh TensorFlow (ditentukan oleh kunci tfrecord+tfe ). 

import tensorflow_decision_forests as tfdf
import tensorflow as tf

strategy = tf.distribute.experimental.ParameterServerStrategy(...)

with strategy.scope():
  model = tfdf.keras.DistributedGradientBoostedTreesModel()

model.fit_on_dataset_path(
    train_path="/path/to/dataset/train@1000",
    label_key="label_key",
    dataset_format="tfrecord+tfe")

print("Trained model")
model.summary()

Contoh: Pelatihan terdistribusi pada kumpulan data terdistribusi TensorFlow yang terbatas

TF-DF mengharapkan kumpulan data TensorFlow dengan pecahan pekerja terbatas yang terdistribusi:

  • Terdistribusi : Kumpulan data yang tidak terdistribusi dibungkus dalam strategy.distribute_datasets_from_function .
  • finite : Kumpulan data harus membaca setiap contoh tepat satu kali. Kumpulan data tidak boleh berisi instruksi repeat apa pun.
  • sharded pekerja : Setiap pekerja harus membaca bagian terpisah dari kumpulan data.

Berikut ini contohnya: 

import tensorflow_decision_forests as tfdf
import tensorflow as tf


def dataset_fn(context, paths):
  """Create a worker-sharded finite dataset from paths.

  Like for non-distributed training, each example should be visited exactly
  once (and by only one worker) during the training. In addition, for optimal
  training speed, the reading of the examples should be distributed among the
  workers (instead of being read by a single worker, or read and discarded
  multiple times).

  In other words, don't add a "repeat" statement and make sure to shard the
  dataset at the file level and not at the example level.
  """

  # List the dataset files
  ds_path = tf.data.Dataset.from_tensor_slices(paths)

  # Make sure the dataset is used with distributed training.
  assert context is not None


  # Split the among the workers.
  #
  # Note: The "shard" is applied on the file path. The shard should not be
  # applied on the examples directly.
  # Note: You cannot use 'context.num_input_pipelines' with ParameterServerV2.
  current_worker = tfdf.keras.get_worker_idx_and_num_workers(context)
  ds_path = ds_path.shard(
      num_shards=current_worker.num_workers,
      index=current_worker.worker_idx)

  def read_csv_file(path):
    """Reads a single csv file."""

    numerical = tf.constant([0.0], dtype=tf.float32)
    categorical_string = tf.constant(["NA"], dtype=tf.string)
    csv_columns = [
        numerical,  # feature 1
        categorical_string,  # feature 2
        numerical,  # feature 3
        # ... define the features here.
    ]
    return tf.data.experimental.CsvDataset(path, csv_columns, header=True)

  ds_columns = ds_path.interleave(read_csv_file)

  # We assume a binary classification label with the following possible values.
  label_values = ["<=50K", ">50K"]

  # Convert the text labels into integers:
  # "<=50K" => 0
  # ">50K" => 1
  init_label_table = tf.lookup.KeyValueTensorInitializer(
      keys=tf.constant(label_values),
      values=tf.constant(range(label_values), dtype=tf.int64))
  label_table = tf.lookup.StaticVocabularyTable(
      init_label_table, num_oov_buckets=1)

  def extract_label(*columns):
    return columns[0:-1], label_table.lookup(columns[-1])

  ds_dataset = ds_columns.map(extract_label)

  # The batch size has no impact on the quality of the model. However, a larger
  # batch size generally is faster.
  ds_dataset = ds_dataset.batch(500)
  return ds_dataset


strategy = tf.distribute.experimental.ParameterServerStrategy(...)
with strategy.scope():
  model = tfdf.keras.DistributedGradientBoostedTreesModel()

  train_dataset = strategy.distribute_datasets_from_function(
      lambda context: dataset_fn(context, [...list of csv files...])
  )

model.fit(train_dataset)

print("Trained model")
model.summary()

Contoh: Penyetelan hyper-parameter terdistribusi pada jalur himpunan data

Penyetelan hyper-parameter terdistribusi pada jalur himpunan data mirip dengan pelatihan terdistribusi. Satu-satunya perbedaan adalah opsi ini kompatibel dengan model yang tidak terdistribusi. Misalnya, Anda dapat mendistribusikan penyetelan hyper-parameter model Gradient Boosted Trees (yang tidak terdistribusi). 

with strategy.scope():
  tuner = tfdf.tuner.RandomSearch(num_trials=30, use_predefined_hps=True)
  model = tfdf.keras.GradientBoostedTreesModel(tuner=tuner)

training_history = model.fit_on_dataset_path(
  train_path=train_path,
  label_key=label,
  dataset_format="csv",
  valid_path=test_path)

logging.info("Trained model:")
model.summary()

Contoh: Pengujian unit

Untuk menguji unit pelatihan terdistribusi, Anda dapat membuat proses pekerja tiruan. Lihat metode _create_in_process_tf_ps_cluster dalam pengujian unit TF-DF untuk informasi lebih lanjut.