Aracılığıyla paylaş

kullanarak XGBoost modellerinin dağıtılmış eğitimi xgboost.spark

  • Makale

Önemli

Bu özellik Genel Önizlemededir.

xgboost>=1.7 Python paketi yeni bir modül xgboost.sparkiçerir. Bu modül xgboost PySpark tahmin aracılarını xgboost.spark.SparkXGBRegressor, xgboost.spark.SparkXGBClassifierve xgboost.spark.SparkXGBRankeriçerir. Bu yeni sınıflar, SparkML İşlem Hatlarına XGBoost tahmin aracılarının eklenmesini destekler. API ayrıntıları için bkz . XGBoost python spark API belgesi.

Gereksinimler

Databricks Runtime 12.0 ML ve üzeri.

xgboost.spark Parametre

Modülde tanımlanan tahmin araçları, standart XGBoost'ta xgboost.spark kullanılan parametrelerin ve bağımsız değişkenlerin çoğunu destekler.

  • Sınıf oluşturucusunun, fit yönteminin ve predict yönteminin parametreleri modüldekilerle xgboost.sklearn büyük ölçüde aynıdır.
  • Adlandırma, değerler ve varsayılanlar çoğunlukla XGBoost parametrelerinde açıklananlarla aynıdır.
  • Özel durumlar, desteklenmeyen birkaç parametredir (, gpu_idnthread, , sample_weightgibieval_set) ve pyspark eklenmiş tahmin aracıya özgü parametrelerdir (örneğinfeaturesCol, , labelCol, use_gpu). validationIndicatorCol Ayrıntılar için bkz . XGBoost Python Spark API belgeleri.

Dağıtılmış eğitim

Modülde xgboost.spark tanımlanan PySpark tahmin araçları, parametresini num_workers kullanarak dağıtılmış XGBoost eğitimini destekler. Dağıtılmış eğitimi kullanmak için bir sınıflandırıcı veya regresör oluşturun ve dağıtılmış eğitim sırasında eşzamanlı çalışan Spark görevlerinin sayısına ayarlayın num_workers . Tüm Spark görev yuvalarını kullanmak için öğesini ayarlayın num_workers=sc.defaultParallelism.

Örneğin:

from xgboost.spark import SparkXGBClassifier
classifier = SparkXGBClassifier(num_workers=sc.defaultParallelism)

Not

  • Dağıtılmış XGBoost ile kullanamazsınız mlflow.xgboost.autolog . MLflow kullanarak xgboost Spark modelini günlüğe kaydetmek için kullanın mlflow.spark.log_model(spark_xgb_model, artifact_path).
  • Otomatik ölçeklendirmenin etkinleştirildiği bir kümede dağıtılmış XGBoost kullanamazsınız. Bu esnek ölçeklendirme paradigmasıyla başlayan yeni çalışan düğümleri yeni görev kümeleri alamaz ve boşta kalır. Otomatik ölçeklendirmeyi devre dışı bırakma yönergeleri için bkz . Otomatik ölçeklendirmeyi etkinleştirme.

Seyrek özellikler veri kümesi üzerinde eğitim için iyileştirmeyi etkinleştirme

Modülde xgboost.spark tanımlanan PySpark Tahmin Araçları, seyrek özelliklere sahip veri kümeleri üzerinde eğitim için iyileştirmeyi destekler. Seyrek özellik kümelerinin en iyi duruma getirilmesini sağlamak için, türündeki pyspark.ml.linalg.SparseVector değerlerden oluşan özellikler sütununu içeren yönteme fit bir veri kümesi sağlamanız ve tahmin aracı parametresini enable_sparse_data_optim olarak Trueayarlamanız gerekir. Ayrıca parametresini missing olarak 0.0ayarlamanız gerekir.

Örneğin:

from xgboost.spark import SparkXGBClassifier
classifier = SparkXGBClassifier(enable_sparse_data_optim=True, missing=0.0)
classifier.fit(dataset_with_sparse_features_col)

GPU eğitimi

Modülde xgboost.spark tanımlanan PySpark tahmin araçları GPU'larla ilgili eğitimi destekler. GPU eğitimini etkinleştirmek için parametresini use_gpu olarak True ayarlayın.

Not

XGBoost dağıtılmış eğitiminde kullanılan her Spark görevi için, bağımsız değişken olarak ayarlandığında Trueeğitimde use_gpu yalnızca bir GPU kullanılır. Databricks, Spark kümesi yapılandırması spark.task.resource.gpu.amountiçin varsayılan değerinin 1 kullanılmasını önerir. Aksi takdirde, bu Spark görevine ayrılan ek GPU'lar boşta olur.

Örneğin:

from xgboost.spark import SparkXGBClassifier
classifier = SparkXGBClassifier(num_workers=sc.defaultParallelism, use_gpu=True)

Sorun giderme

Çok düğümlü eğitim sırasında bir NCCL failure: remote process exited or there was a network error iletiyle karşılaşırsanız bu durum genellikle GPU'lar arasındaki ağ iletişimi ile ilgili bir sorun olduğunu gösterir. Bu sorun, NCCL (NVIDIA Kolektif İletişim Kitaplığı) GPU iletişimi için belirli ağ arabirimlerini kullanamadığında ortaya çıkar.

Sorunu çözmek için, için kümenin sparkConf spark.executorEnv.NCCL_SOCKET_IFNAME değerini olarak ethayarlayın. Bu temelde bir düğümdeki tüm çalışanlar için ortam değişkenini NCCL_SOCKET_IFNAMEeth olarak ayarlar.

Örnek not defteri

Bu not defteri, Spark MLlib ile Python paketinin xgboost.spark kullanımını gösterir.

PySpark-XGBoost not defteri

Not defterini alma

Kullanım dışı bırakılan sparkdl.xgboost modül için geçiş kılavuzu

  • ile from xgboost.spark import SparkXGBRegressor değiştirin from sparkdl.xgboost import XgboostRegressor ve değerini ile from xgboost.spark import SparkXGBClassifierdeğiştirinfrom sparkdl.xgboost import XgboostClassifier.
  • Tahmin aracı oluşturucusunda camelCase stilindeki tüm parametre adlarını snake_case stil olarak değiştirin. Örneğin, olarak SparkXGBRegressor(features_col=XXX)değiştirinXgboostRegressor(featuresCol=XXX).
  • ve parametreleri use_external_storageexternal_storage_precision kaldırıldı. xgboost.spark tahmin araçları, belleği daha verimli kullanmak için DMatrix veri yineleme API'sini kullanır. Artık verimsiz dış depolama modunu kullanmaya gerek yoktur. Son derece büyük veri kümeleri için Databricks, her eğitim görevinin num_workers verileri daha küçük, daha yönetilebilir veri bölümlerine ayırmasını sağlayan parametresini artırmanızı önerir. kümedeki toplam Spark görev yuvası sayısına ayarlayan num_workers ayarını num_workers = sc.defaultParallelismgöz önünde bulundurun.
  • içinde tanımlanan tahmin araçları için xgboost.sparkayarı num_workers=1 , tek bir Spark görevi kullanarak model eğitimini yürütür. Bu, varsayılan olarak 1 olan Spark kümesi yapılandırma ayarı spark.task.cpustarafından belirtilen CPU çekirdeği sayısını kullanır. Modeli eğitmek için daha fazla CPU çekirdeği kullanmak için veya spark.task.cpusöğesini artırınnum_workers. içinde xgboost.sparktanımlanan tahmin araçları için veya n_jobs parametresini ayarlayamazsınıznthread. Bu davranış, kullanım sparkdl.xgboost dışı bırakılmış pakette tanımlanan tahmincilerin önceki davranışından farklıdır.

Modeli modele dönüştürme sparkdl.xgboostxgboost.spark

sparkdl.xgboostmodeller modellerden xgboost.spark farklı bir biçimde kaydedilir ve farklı parametre ayarlarına sahiptir. Modeli dönüştürmek için aşağıdaki yardımcı program işlevini kullanın:

def convert_sparkdl_model_to_xgboost_spark_model(
  xgboost_spark_estimator_cls,
  sparkdl_xgboost_model,
):
  """
  :param xgboost_spark_estimator_cls:
      `xgboost.spark` estimator class, e.g. `xgboost.spark.SparkXGBRegressor`
  :param sparkdl_xgboost_model:
      `sparkdl.xgboost` model instance e.g. the instance of
       `sparkdl.xgboost.XgboostRegressorModel` type.

  :return
      A `xgboost.spark` model instance
  """

  def convert_param_key(key):
    from xgboost.spark.core import _inverse_pyspark_param_alias_map
    if key == "baseMarginCol":
      return "base_margin_col"
    if key in _inverse_pyspark_param_alias_map:
      return _inverse_pyspark_param_alias_map[key]
    if key in ['use_external_storage', 'external_storage_precision', 'nthread', 'n_jobs', 'base_margin_eval_set']:
      return None
    return key

  xgboost_spark_params_dict = {}
  for param in sparkdl_xgboost_model.params:
    if param.name == "arbitraryParamsDict":
      continue
    if sparkdl_xgboost_model.isDefined(param):
      xgboost_spark_params_dict[param.name] = sparkdl_xgboost_model.getOrDefault(param)

  xgboost_spark_params_dict.update(sparkdl_xgboost_model.getOrDefault("arbitraryParamsDict"))

  xgboost_spark_params_dict = {
    convert_param_key(k): v
    for k, v in xgboost_spark_params_dict.items()
    if convert_param_key(k) is not None
  }

  booster = sparkdl_xgboost_model.get_booster()
  booster_bytes = booster.save_raw("json")
  booster_config = booster.save_config()
  estimator = xgboost_spark_estimator_cls(**xgboost_spark_params_dict)
  sklearn_model = estimator._convert_to_sklearn_model(booster_bytes, booster_config)
  return estimator._copyValues(estimator._create_pyspark_model(sklearn_model))

# Example
from xgboost.spark import SparkXGBRegressor

new_model = convert_sparkdl_model_to_xgboost_spark_model(
  xgboost_spark_estimator_cls=SparkXGBRegressor,
  sparkdl_xgboost_model=model,
)

Son aşama olarak model içeren bir pyspark.ml.PipelineModelsparkdl.xgboost modeliniz varsa, modelin aşamasını sparkdl.xgboost dönüştürülmüş xgboost.spark modelle değiştirebilirsiniz.

pipeline_model.stages[-1] = convert_sparkdl_model_to_xgboost_spark_model(
  xgboost_spark_estimator_cls=SparkXGBRegressor,
  sparkdl_xgboost_model=pipeline_model.stages[-1],
)