网站设计学习,东莞常平隐贤山庄门票多少钱,紧固件网站建设,静态网页代码大全文章目录 1、分布式计算场景#xff08;1#xff09;分类#xff08;2#xff09;回归#xff08;3#xff09;时间序列 2、分布式应用技巧(1)一个更实际的案例(2) 在设置中使用lambda而不是dataframe(3) 保持确定性(4) 设置n_jobs(4)设置适当的批量大小(5) 显示进度(6)自… 文章目录 1、分布式计算场景1分类2回归3时间序列 2、分布式应用技巧(1)一个更实际的案例(2) 在设置中使用lambda而不是dataframe(3) 保持确定性(4) 设置n_jobs(4)设置适当的批量大小(5) 显示进度(6)自定义指标(7) Spark设置(8) Dask(9) 本地并行化(10) 如何开发 Fugue 是一个低代码的统一接口用于不同的计算框架如 Spark、Dask。PyCaret 使用 Fugue 来支持分布式计算场景。 1、分布式计算场景
1分类
让我们从最标准的例子开始代码与本地版本完全相同没有任何魔法。
# 导入所需的库
from pycaret.datasets import get_data # 导入获取数据的函数
from pycaret.classification import * # 导入分类模型# 使用get_data函数获取名为juice的数据集并设置verbose参数为False表示不显示详细信息
data get_data(juice, verboseFalse)# 设置目标变量为Purchasen_jobs参数为1表示使用单个进程
setup(datadata, targetPurchase, n_jobs1)# 获取前5个模型的名称并存储在test_models变量中
test_models models().index.tolist()[:5]compare_model如果您不想使用分布式系统也完全相同。
# 比较模型函数
compare_models(includetest_models, n_select2)Processing: 0%| | 0/26 [00:00?, ?it/s][LogisticRegression(C1.0, class_weightNone, dualFalse, fit_interceptTrue,intercept_scaling1, l1_ratioNone, max_iter1000,multi_classauto, n_jobsNone, penaltyl2,random_state4292, solverlbfgs, tol0.0001, verbose0,warm_startFalse),DecisionTreeClassifier(ccp_alpha0.0, class_weightNone, criteriongini,max_depthNone, max_featuresNone, max_leaf_nodesNone,min_impurity_decrease0.0, min_samples_leaf1,min_samples_split2, min_weight_fraction_leaf0.0,random_state4292, splitterbest)]现在让我们将其分布式作为一个玩具案例在dask上。唯一改变的是一个额外的参数parallel_backend。
# 导入所需的库
from pycaret.parallel import FugueBackend# 使用FugueBackend作为并行计算的后端# compare_models函数用于比较多个模型的性能
# include参数指定要比较的模型列表
# n_select参数指定要选择的最佳模型数量
# parallel参数指定使用的并行计算后端这里使用FugueBackend(dask)表示使用Dask作为并行计算后端compare_models(includetest_models, n_select2, parallelFugueBackend(dask))[LogisticRegression(C1.0, class_weightNone, dualFalse, fit_interceptTrue,intercept_scaling1, l1_ratioNone, max_iter1000,multi_classauto, n_jobsNone, penaltyl2,random_state4292, solverlbfgs, tol0.0001, verbose0,warm_startFalse),DecisionTreeClassifier(ccp_alpha0.0, class_weightNone, criteriongini,max_depthNone, max_featuresNone, max_leaf_nodesNone,min_impurity_decrease0.0, min_samples_leaf1,min_samples_split2, min_weight_fraction_leaf0.0,random_state4292, splitterbest)]为了使用Spark作为执行引擎您必须能够访问一个Spark集群并且必须拥有一个SparkSession让我们初始化一个本地的Spark会话。
# 导入SparkSession模块
from pyspark.sql import SparkSession# 创建或获取SparkSession对象
spark SparkSession.builder.getOrCreate()现在只需将parallel_backend更改为此会话对象即可在Spark上运行。您必须明白这只是一个玩具案例。在实际情况中您需要拥有一个指向真实Spark集群的SparkSession才能享受Spark的强大功能。
# 调用 compare_models 函数传入参数 includetest_models、n_select2 和 parallelFugueBackend(spark)
compare_models(includetest_models, n_select2, parallelFugueBackend(spark))[LogisticRegression(C1.0, class_weightNone, dualFalse, fit_interceptTrue,intercept_scaling1, l1_ratioNone, max_iter1000,multi_classauto, n_jobsNone, penaltyl2,random_state4292, solverlbfgs, tol0.0001, verbose0,warm_startFalse),DecisionTreeClassifier(ccp_alpha0.0, class_weightNone, criteriongini,max_depthNone, max_featuresNone, max_leaf_nodesNone,min_impurity_decrease0.0, min_samples_leaf1,min_samples_split2, min_weight_fraction_leaf0.0,random_state4292, splitterbest)]最后你可以使用pull命令来获取指标表格。
pull()2回归
回归问题与分类问题遵循相同的模式。
# 导入所需的库
from pycaret.datasets import get_data # 导入获取数据的函数
from pycaret.regression import * # 导入回归模型# 设置数据和目标变量
setup(dataget_data(insurance, verboseFalse), targetcharges, n_jobs1)# 获取前5个模型
test_models models().index.tolist()[:5]compare_model如果您不想使用分布式系统也完全相同。
# 比较模型性能的函数
# 参数
# include: 需要比较的模型列表
# n_select: 需要选择的模型数量
# sort: 按照哪个指标进行排序默认为平均绝对误差MAEcompare_models(includetest_models, n_select2, sortMAE)Processing: 0%| | 0/26 [00:00?, ?it/s][Lars(copy_XTrue, eps2.220446049250313e-16, fit_interceptTrue, fit_pathTrue,jitterNone, n_nonzero_coefs500, normalizedeprecated,precomputeauto, random_state3514, verboseFalse),LinearRegression(copy_XTrue, fit_interceptTrue, n_jobs1,normalizedeprecated, positiveFalse)]现在让我们将其分布式作为一个玩具案例在dask上。唯一改变的是一个额外的参数parallel_backend。
# 导入所需的库
from pycaret.parallel import FugueBackend# 使用FugueBackend作为并行计算的后端# compare_models函数用于比较多个模型的性能并选择性能最好的几个模型
# include参数指定要比较的模型列表
# n_select参数指定要选择的模型数量
# sort参数指定按照哪个指标进行排序这里选择按照平均绝对误差MAE进行排序
# parallel参数指定使用的并行计算后端这里选择使用FugueBackend(dask)作为并行计算的后端
compare_models(includetest_models, n_select2, sortMAE, parallelFugueBackend(dask))[Lars(copy_XTrue, eps2.220446049250313e-16, fit_interceptTrue, fit_pathTrue,jitterNone, n_nonzero_coefs500, normalizedeprecated,precomputeauto, random_state3514, verboseFalse),LinearRegression(copy_XTrue, fit_interceptTrue, n_jobs1,normalizedeprecated, positiveFalse)]为了使用Spark作为执行引擎您必须能够访问一个Spark集群并且必须拥有一个SparkSession让我们初始化一个本地的Spark会话。
# 导入SparkSession模块
from pyspark.sql import SparkSession# 创建或获取一个SparkSession对象
spark SparkSession.builder.getOrCreate()现在只需将parallel_backend更改为此会话对象即可在Spark上运行。您必须明白这只是一个玩具案例。在真实情况下您需要拥有一个指向真实Spark集群的SparkSession才能享受Spark的强大功能。
# 调用compare_models函数传入参数includetest_models、n_select2、sortMAE和parallelFugueBackend(spark)
compare_models(includetest_models, n_select2, sortMAE, parallelFugueBackend(spark))[Lars(copy_XTrue, eps2.220446049250313e-16, fit_interceptTrue, fit_pathTrue,jitterNone, n_nonzero_coefs500, normalizedeprecated,precomputeauto, random_state3514, verboseFalse),LinearRegression(copy_XTrue, fit_interceptTrue, n_jobs1,normalizedeprecated, positiveFalse)]最后你可以使用pull命令来获取指标表格。
pull()3时间序列
它遵循与分类相同的模式。
# 导入所需的库和模块
from pycaret.datasets import get_data # 导入获取数据的函数
from pycaret.time_series import * # 导入时间序列模块# 创建时间序列预测实验对象
exp TSForecastingExperiment()# 设置实验参数
exp.setup(dataget_data(airline, verboseFalse), # 获取数据集此处使用航空数据集fh12, # 设置预测的未来时间步数为12fold3, # 设置交叉验证的折数为3fig_kwargs{renderer: notebook}, # 设置绘图参数此处使用notebook作为渲染器session_id42 # 设置随机种子为42保证实验的可重复性
)# 获取前5个模型的名称
test_models exp.models().index.tolist()[:5]# 比较模型性能并选择最佳模型# 使用exp.compare_models函数比较模型性能并选择最佳的3个模型作为基准模型
# 参数includetest_models表示只比较test_models中的模型
# 参数n_select3表示选择性能最好的3个模型作为最佳基准模型
best_baseline_models exp.compare_models(includetest_models, n_select3)
best_baseline_modelsProcessing: 0%| | 0/27 [00:00?, ?it/s][ARIMA(maxiter50, methodlbfgs, order(1, 0, 0), out_of_sample_size0,scoringmse, scoring_argsNone, seasonal_order(0, 1, 0, 12),start_paramsNone, suppress_warningsFalse, trendNone,with_interceptTrue),NaiveForecaster(sp12, strategylast, window_lengthNone),PolynomialTrendForecaster(degree1, regressorNone, with_interceptTrue)]# 导入所需的模块
from pycaret.parallel import FugueBackend# 使用FugueBackend作为并行计算的后端
# FugueBackend是一个用于分布式计算的后端可以使用Dask或Ray来实现并行计算
# 这里使用了Dask作为FugueBackend的参数表示使用Dask来进行并行计算# 使用exp.compare_models函数比较模型性能并选择最佳的3个模型
# include参数指定要比较的模型列表test_models是一个包含待比较模型的列表
# n_select参数指定要选择的最佳模型的数量这里选择了3个最佳模型
# parallel参数指定并行计算的后端这里使用了之前创建的FugueBackend对象# 将比较结果保存在best_baseline_models变量中该变量将包含最佳的3个模型
best_baseline_models exp.compare_models(includetest_models, n_select3, parallelFugueBackend(dask))
best_baseline_models[ARIMA(maxiter50, methodlbfgs, order(1, 0, 0), out_of_sample_size0,scoringmse, scoring_argsNone, seasonal_order(0, 1, 0, 12),start_paramsNone, suppress_warningsFalse, trendNone,with_interceptTrue),NaiveForecaster(sp12, strategylast, window_lengthNone),PolynomialTrendForecaster(degree1, regressorNone, with_interceptTrue)]# 导入SparkSession模块
from pyspark.sql import SparkSession# 创建或获取SparkSession对象
spark SparkSession.builder.getOrCreate()# 导入所需的模块
from pycaret.parallel import FugueBackend# 使用FugueBackend作为并行计算的后端# 使用exp.compare_models函数来比较模型性能并选择最佳模型
# include参数指定要比较的模型列表这里选择了test_models列表的前两个模型
# n_select参数指定要选择的最佳模型数量这里选择了3个最佳模型
# parallel参数指定并行计算的后端这里使用了FugueBackend(spark)# 将比较结果保存在best_baseline_models变量中
best_baseline_models exp.compare_models(includetest_models[:2], n_select3, parallelFugueBackend(spark))
best_baseline_models[NaiveForecaster(sp1, strategylast, window_lengthNone),NaiveForecaster(sp1, strategymean, window_lengthNone)]# 从exp对象中调用pull()方法
exp.pull()2、分布式应用技巧
(1)一个更实际的案例
上面的例子都是纯粹的玩具为了在分布式系统中使事情完美运行你必须注意一些事情
(2) 在设置中使用lambda而不是dataframe
如果你直接在setup中提供一个dataframe这个数据集将需要发送到所有的工作节点。如果dataframe是1G你有100个工作节点那么你的驱动机器可能需要发送高达100G的数据取决于具体框架的实现这个数据传输本身就成为了一个瓶颈。相反如果你提供一个lambda函数它不会改变本地计算的情况但驱动程序只会将函数引用发送给工作节点每个工作节点将负责自己加载数据因此驱动程序端没有大量的流量。
(3) 保持确定性
你应该始终使用session_id来使分布式计算具有确定性。
(4) 设置n_jobs
在想要分布式运行某些任务时明确设置n_jobs非常重要这样它就不会过度使用本地/远程资源。这也可以避免资源争用并加快计算速度。
# 导入所需的库
from pycaret.datasets import get_data # 导入获取数据的函数
from pycaret.classification import * # 导入分类模块# 设置函数用于获取数据
# 使用get_data函数获取名为juice的数据集关闭冗长输出(verboseFalse)关闭数据集的概要信息(profileFalse)
# 设置目标变量为Purchase
# 设置会话ID为0以确保结果的可重复性
# 设置使用的CPU核心数为1
setup(data_funclambda: get_data(juice, verboseFalse, profileFalse), targetPurchase, session_id0, n_jobs1);(4)设置适当的批量大小
batch_size参数有助于在负载均衡和开销之间进行调整。对于每个批次设置将只调用一次。所以
选择负载均衡开销最佳情况较小的批量大小更好更差训练时间 数据加载时间 或者 模型数量 ~ 工作进程数量较大的批量大小更差更好训练时间 数据加载时间 或者 模型数量 工作进程数量
默认值设置为1表示我们希望获得最佳的负载均衡。
(5) 显示进度
在开发中您可以通过display_remoteTrue启用可视效果但同时您还必须启用Fugue回调以便驱动程序可以监视工作进度。但建议在生产环境中关闭显示。
# 导入所需的模块
from pycaret.parallel import FugueBackend# 定义配置参数
fconf {fugue.rpc.server: fugue.rpc.flask.FlaskRPCServer, # 保持该值不变fugue.rpc.flask_server.host: 0.0.0.0, # 驱动程序的 IP 地址工作节点可以访问fugue.rpc.flask_server.port: 3333, # 驱动程序上的开放端口fugue.rpc.flask_server.timeout: 2 sec, # 工作节点与驱动程序通信的超时时间
}# 创建 FugueBackend 对象
be FugueBackend(dask, fconf, display_remoteTrue, batch_size3, top_onlyFalse)# 使用 FugueBackend 对象进行模型比较
compare_models(n_select2, parallelbe)Processing: 0%| | 0/14 [00:00?, ?it/s][RidgeClassifier(alpha1.0, class_weightNone, copy_XTrue, fit_interceptTrue,max_iterNone, normalizedeprecated, positiveFalse,random_state0, solverauto, tol0.001),LinearDiscriminantAnalysis(covariance_estimatorNone, n_componentsNone,priorsNone, shrinkageNone, solversvd,store_covarianceFalse, tol0.0001)](6)自定义指标
您可以像以前一样添加自定义指标。但是为了使评分器可分发它必须是可序列化的。一个常见的函数应该没问题但是如果在函数内部使用了一些不可序列化的全局变量例如一个RLock对象可能会引发问题。因此请尽量使自定义函数独立于全局变量。
# 定义一个名为score_dummy的函数用于计算模型的得分
# 参数y_true表示真实值y_pred表示预测值axis表示计算得分的轴
def score_dummy(y_true, y_pred, axis0):return 0.0# 添加一个名为mydummy的指标
# 参数id表示指标的唯一标识符
# 参数name表示指标的名称
# 参数score_func表示计算指标得分的函数这里使用之前定义的score_dummy函数
# 参数target表示指标的计算目标这里是预测值
# 参数greater_is_better表示得分是否越大越好这里设置为False表示得分越小越好
add_metric(idmydummy,nameDUMMY,score_funcscore_dummy,targetpred,greater_is_betterFalse)Name DUMMY
Display Name DUMMY
Score Function function score_dummy at 0x7f8aa0dc0ca0
Scorer make_scorer(score_dummy, greater_is_betterFalse)
Target pred
Args {}
Greater is Better False
Multiclass True
Custom True
Name: mydummy, dtype: object在类实例中添加一个函数也是可以的但是请确保类中的所有成员变量都是可序列化的。
# 获取模型列表的前5个模型
test_models models().index.tolist()[:5]# 比较模型
# include参数指定要比较的模型列表
# n_select参数指定要选择的模型数量
# sort参数指定排序方式这里使用DUMMY表示不进行排序
# parallel参数指定使用的并行计算后端这里使用Dask作为后端
compare_models(includetest_models, n_select2, sortDUMMY, parallelFugueBackend(dask))[DecisionTreeClassifier(ccp_alpha0.0, class_weightNone, criteriongini,max_depthNone, max_featuresNone, max_leaf_nodesNone,min_impurity_decrease0.0, min_samples_leaf1,min_samples_split2, min_weight_fraction_leaf0.0,random_state0, splitterbest),LogisticRegression(C1.0, class_weightNone, dualFalse, fit_interceptTrue,intercept_scaling1, l1_ratioNone, max_iter1000,multi_classauto, n_jobsNone, penaltyl2,random_state0, solverlbfgs, tol0.0001, verbose0,warm_startFalse)]pull()# 定义一个Scores类
class Scores:# 定义一个名为score_dummy2的方法用于计算得分# 参数y_true表示真实标签y_prob表示预测概率axis表示轴def score_dummy2(self, y_true, y_prob, axis0):return 1.0# 创建一个Scores对象
scores Scores()# 添加一个指标
add_metric(idmydummy2, # 指标的唯一标识符nameDUMMY2, # 指标的名称score_funcscores.score_dummy2, # 指标的计算函数targetpred_proba, # 指标的目标值这里是预测概率greater_is_betterTrue, # 指标的得分越大越好
)Name DUMMY2
Display Name DUMMY2
Score Function bound method Scores.score_dummy2 of __main__...
Scorer make_scorer(score_dummy2, needs_probaTrue, er...
Target pred_proba
Args {}
Greater is Better True
Multiclass True
Custom True
Name: mydummy2, dtype: object# 调用compare_models函数传入参数includetest_models、n_select2、sortDUMMY2和parallelFugueBackend(dask)
compare_models(includetest_models, n_select2, sortDUMMY2, parallelFugueBackend(dask))[DecisionTreeClassifier(ccp_alpha0.0, class_weightNone, criteriongini,max_depthNone, max_featuresNone, max_leaf_nodesNone,min_impurity_decrease0.0, min_samples_leaf1,min_samples_split2, min_weight_fraction_leaf0.0,random_state0, splitterbest),LogisticRegression(C1.0, class_weightNone, dualFalse, fit_interceptTrue,intercept_scaling1, l1_ratioNone, max_iter1000,multi_classauto, n_jobsNone, penaltyl2,random_state0, solverlbfgs, tol0.0001, verbose0,warm_startFalse)]# 这是一个函数定义函数名为pull
def pull():# 这是一个空函数没有任何代码pass(7) Spark设置
强烈建议每个Spark执行器上只有一个worker这样worker可以充分利用所有的CPU设置spark.task.cpus。当你这样做时你应该明确地在setup中设置n_jobs为每个执行器的CPU数量。
executor_cores 4spark SparkSession.builder.config(spark.task.cpus, executor_cores).config(spark.executor.cores, executor_cores).getOrCreate()setup(dataget_data(juice, verboseFalse, profileFalse), target Purchase, session_id0, n_jobsexecutor_cores)compare_models(n_select2, parallelFugueBackend(spark))(8) Dask
Dask有假分布式模式例如默认的多线程和多进程模式。默认模式可以正常工作但实际上是按顺序运行的而多进程模式目前对PyCaret不起作用因为它会干扰PyCaret的全局变量。另一方面任何Spark执行模式都可以正常工作。
(9) 本地并行化
对于尝试非平凡数据和模型的实际用途本地并行化最简单的方法是使用上面显示的本地Dask作为后端通常没有性能优势。因为在训练过程中很容易超载CPU增加资源争用。本地并行化的价值在于验证代码并让你相信分布式环境将在更短的时间内提供预期的结果。
(10) 如何开发
分布式系统很强大但你必须遵循一些良好的实践来使用它们
从小到大 最初你必须从一小组数据开始例如在compare_model中将你想尝试的模型限制为一小组廉价模型当你验证它们工作正常后可以切换到更大的模型集合。从本地到分布式 你应该按照这个顺序进行先在本地验证小数据然后在分布式环境下验证小数据最后在分布式环境下验证大数据。当前的设计使过渡无缝。你可以按顺序进行这些操作parallelNone - parallelFugueBackend() - parallelFugueBackend(spark)。在第二步中你可以替换为本地的SparkSession或本地的dask。
