PredictionIO 批量持久化评估器实战：用 `pio eval` 为一组查询批量产出推荐预测结果

PredictionIO 批量持久化评估器实战用pio eval为一组查询批量产出推荐预测结果【免费下载链接】predictionioPredictionIO, a machine learning server for developers and ML engineers.项目地址: https://gitcode.com/gh_mirrors/pred/predictionio本文基于 Apache PredictionIO 的 Recommendation 模板v0.3.2完整讲解如何改造 DataSource 的readEval()生成批量查询、编写一个不做指标计算而是把 (Query, PredictedResult) 直接落盘的BatchPersistableEvaluator并配合Evaluation与EngineParamsGenerator通过一条pio eval命令输出批量预测文件。读完本文你将掌握如何为任意一批用户/参数组合批量执行推荐预测、如何把结果以 JSON 文本文件形式持久化、以及该方案在源码层的运行链路BaseEvaluator、Evaluation、CoreWorkflow.runEvaluation。适用前提与阅读前置本文对应的模板版本为Recommendation template v0.3.2。文中使用$pio eval配合自定义 Evaluator 将一组查询的预测结果持久化到输出目录属于实验性/开发者特性experimental and developer features未来版本可能发生变更。在动手之前建议先阅读 Evaluation Explained (Recommendation)理解两件事DataSource 中readEval()的职责——返回Seq[(TrainingData, EmptyEvaluationInfo, RDD[(Query, ActualResult)])]即训练数据、空的评估信息、以及「查询 → 实际结果」的配对 RDDEvaluation 组件的用法——Evaluation定义了「用哪个引擎 哪个评估器」跑评估。常规的推荐评估如 evaluation.html.md.erb 中介绍的MetricEvaluatorPrecisionAtK会计算准确率等指标并把最优参数与指标分数打印出来。而本文要做的恰恰相反不关心指标分数只关心「给定一批查询每个查询会得到什么推荐结果」——这正是批量离线预测batch predict的典型场景。整体思路整个方案由三部分组成缺一不可组成文件建议命名作用改造后的 DataSourceDataSource.scala在readEval()中构造我们想要批量预测的Query列表并配上 dummy 的ActualResult自定义 EvaluatorBatchPersistableEvaluator.scala继承BaseEvaluator接收评估流水线给出的 (Query, PredictedResult, ActualResult) RDD序列化为 JSON 后落盘评估入口对象BatchEvaluation.scala定义Evaluation绑定新 Evaluator与EngineParamsGenerator指定引擎参数供pio eval调用它们的配合方式pio eval Evaluation EngineParamsGenerator启动评估工作流 → 工作流用EngineParams实例化引擎 → 引擎调用 DataSource 的readEval()得到批量查询 → 训练完成后对每个查询做预测 → 把三元组 RDD 交给 Evaluator → Evaluator 写入输出目录。第 1 步改造 DataSource 生成批量查询1.1 覆写readEval()在你模板的DataSource.scala中把readEval()改为返回一批你希望批量预测的查询。下面这段代码是文档给出的示例实现override def readEval(sc: SparkContext) : Seq[(TrainingData, EmptyEvaluationInfo, RDD[(Query, ActualResult)])] { // This function only return one evaluation data set // Create your own queries here. Below are provided as examples. // for example, you may get all distinct user id from the trainingData to create the Query val batchQueries: RDD[Query] sc.parallelize( Seq( Query(user 1, num 10), Query(user 3, num 15), Query(user 5, num 20) ) ) val queryAndActual: RDD[(Query, ActualResult)] batchQueries.map (q // the ActualResult contain dummy empty rating array // because we not interested in Actual result for batch predict purpose. (q, ActualResult(Array())) ) val evalDataSet ( readTraining(sc), new EmptyEvaluationInfo(), queryAndActual ) Seq(evalDataSet) }要点解读返回值结构与常规评估一致仍然是Seq[(TrainingData, EmptyEvaluationInfo, RDD[(Query, ActualResult)])]因此评估工作流无需任何额外适配即可消费它查询内容完全自定义示例用sc.parallelize(Seq(...))硬编码了 3 个查询用户 1 取 10 个推荐、用户 3 取 15 个、用户 5 取 20 个。文档注释指出更常见的做法是从训练数据中取出所有去重后的 user id 来构造查询例如把readTraining得到的TrainingData中的用户集合映射为Query列表ActualResult只放占位数据因为批量预测不关心真实结果这里用ActualResult(Array())填充空的评分数组即可它的作用是让三元组类型完整、通过评估流水线的类型检查。1.2 对比常规评估的readEval()长什么样为了理解上面的改动可以对照 customize-serving 示例的 DataSource.scala。常规实现的readEval()是一个 k-fold 切分过程用getRatings(sc).zipWithUniqueId给每条评分打上唯一 id然后按idx % kFold把数据划分成训练集与测试集再groupBy(_.user)为每个用户构造一条查询并携带该用户在验证集中真实的ActualResult(ratings.toArray)。对比两者可以看出核心差异常规评估查询数量 验证集用户数ActualResult是真实评分用于计算PrecisionK等指标批量预测查询数量与内容完全由你指定ActualResult是空占位评估器完全忽略它。1.3 备选做法新建一个 DataSource 子类文档特别提示也可以不修改原有 DataSource而是新建一个继承原 DataSource 的类来覆写readEval()。这样原始模板代码保持不动只在需要跑批量预测时切换数据源。具体步骤为新建子类并覆写readEval()在Engine.scala中把该子类注册进Engine例如new Engine(classOf[BatchDataSource], ...)在engine.json中指定使用该 Engine 配置。文档原处标注了 “TODO add more details”即这一做法在文档中属于提示性内容具体注册细节可参考引擎默认配置自行扩展。第 2 步编写BatchPersistableEvaluator2.1 为什么需要一个新的 EvaluatorPredictionIO 默认的MetricEvaluator会计算指标分数并把结果写进数据库其工作方式见 Evaluation.scalaengineMetric_会把Metric包装成MetricEvaluator。而我们不需要任何指标计算只需要把「查询 预测结果」原样写盘因此要新建一个 Evaluator。新建文件BatchPersistableEvaluator.scala完整代码如下package org.template.recommendation import org.apache.predictionio.controller.EmptyEvaluationInfo import org.apache.predictionio.controller.Engine import org.apache.predictionio.controller.EngineParams import org.apache.predictionio.controller.EngineParamsGenerator import org.apache.predictionio.controller.Evaluation import org.apache.predictionio.controller.Params import org.apache.predictionio.core.BaseEvaluator import org.apache.predictionio.core.BaseEvaluatorResult import org.apache.predictionio.workflow.WorkflowParams import org.apache.spark.SparkContext import org.apache.spark.rdd.RDD import org.json4s.DefaultFormats import org.json4s.Formats import org.json4s.native.Serialization import grizzled.slf4j.Logger class BatchPersistableEvaluatorResult extends BaseEvaluatorResult {} class BatchPersistableEvaluator extends BaseEvaluator[ EmptyEvaluationInfo, Query, PredictedResult, ActualResult, BatchPersistableEvaluatorResult] { transient lazy val logger Logger[this.type] // A helper object for the json4s serialization case class Row(query: Query, predictedResult: PredictedResult) extends Serializable def evaluateBase( sc: SparkContext, evaluation: Evaluation, engineEvalDataSet: Seq[( EngineParams, Seq[(EmptyEvaluationInfo, RDD[(Query, PredictedResult, ActualResult)])])], params: WorkflowParams): BatchPersistableEvaluatorResult { /** Extract the first data, as we are only interested in the first * evaluation. It is possible to relax this restriction, and have the * output logic below to write to different directory for different engine * params. */ require( engineEvalDataSet.size 1, There should be only one engine params) val evalDataSet engineEvalDataSet.head._2 require(evalDataSet.size 1, There should be only one RDD[(Q, P, A)]) val qpaRDD evalDataSet.head._2 // qpaRDD contains 3 queries we specified in readEval, the corresponding // predictedResults, and the dummy actual result. /** The output directory. Better to use absolute path if you run on cluster. * If your database has a Hadoop interface, you can also convert the * following to write to your database in parallel as well. */ val outputDir batch_result logger.info(Writing result to disk) qpaRDD .map { case (q, p, a) Row(q, p) } .map { row // Convert into a json implicit val formats: Formats DefaultFormats Serialization.write(row) } .saveAsTextFile(outputDir) logger.info(sResult can be found in $outputDir) new BatchPersistableEvaluatorResult() } }2.2 逐段理解这个 Evaluator类型参数BaseEvaluator[EmptyEvaluationInfo, Query, PredictedResult, ActualResult, BatchPersistableEvaluatorResult]。对照 BaseEvaluator.scala 的定义五个类型参数依次是评估信息类EI、查询类Q、预测结果类P、实际结果类A、评估结果类ER。这里EI用EmptyEvaluationInfoER是自定义的BatchPersistableEvaluatorResult继承BaseEvaluatorResult。evaluateBase方法这是BaseEvaluator中唯一需要实现的方法。它的入参中engineEvalDataSet是Seq[(EngineParams, Seq[(EI, RDD[(Q, P, A)])])]——外层对应一组引擎参数内层对应一组评估数据集最内层的RDD[(Q, P, A)]就是「查询、预测结果、实际结果」的三元组 RDD。严格约束输入规模代码里有两个requireengineEvalDataSet.size 1只允许一组引擎参数。因为本示例只为单套参数如rank10, numIterations20, lambda0.01输出一个结果目录若传多组参数会直接抛异常evalDataSet.size 1只允许一个 RDD。因为我们在readEval()中只Seq(evalDataSet)返回了一份数据。文档注释说明如果希望放宽限制可以改造输出逻辑让不同的 engine params 写到不同的目录。落盘逻辑qpaRDD.map { case (q, p, a) Row(q, p) }丢弃不需要的adummy 实际结果只保留Row(query, predictedResult)借助 json4s 的Serialization.write(row)把每条记录序列化为 JSON 字符串。注意implicit val formats: Formats DefaultFormats声明在 map 内部配合import org.json4s.DefaultFormats / Formats / native.Serialization使用saveAsTextFile(outputDir)把整个 RDD 以文本文件形式写入outputDiroutputDir batch_result由局部变量指定在集群上运行时建议改为绝对路径。关于输出目录的扩展saveAsTextFile走的是 Spark 的 Hadoop 文件接口。因此如果存储系统支持 Hadoop 接口可以把同样的逻辑改写成向数据库并行写入见代码注释HDFS 等支持 Hadoop 接口的存储天然可用。2.3 源码依据BaseEvaluator与BaseEvaluatorResultBaseEvaluator.scala 是 PredictionIO 所有评估器的基类被标注为DeveloperApi。关键点evaluateBase(...)由评估工作流Evaluation Workflow调用引擎开发者一般不要直接调用它BaseEvaluatorResult提供toOneLiner()/toHTML()/toJSON()三个默认为空串的方法用于把评估结果呈现到评估 UI以及noSave标志控制结果是否写入数据库。在本文的BatchPersistableEvaluatorResult中这些方法都保持默认空实现因此CoreWorkflow.runEvaluation更新评估实例时拿到的toOneLiner等均为空字符串日志里展示的就是对象默认的toString如org.template.recommendation.BatchPersistableEvaluatorResult2f886889。第 3 步定义Evaluation与EngineParamsGenerator新建文件BatchEvaluation.scala把新 Evaluator 和要使用的引擎参数绑定起来package org.template.recommendation import org.apache.predictionio.controller.EngineParamsGenerator import org.apache.predictionio.controller.EngineParams import org.apache.predictionio.controller.Evaluation object BatchEvaluation extends Evaluation { // Define Engine and Evaluator used in Evaluation /** * Specify the new BatchPersistableEvaluator. */ engineEvaluator (RecommendationEngine(), new BatchPersistableEvaluator()) } object BatchEngineParamsList extends EngineParamsGenerator { // We only interest in a single engine params. engineParamsList Seq( EngineParams( dataSourceParams DataSourceParams(appName INVALID_APP_NAME, evalParams None), algorithmParamsList Seq((als, ALSAlgorithmParams( rank 10, numIterations 20, lambda 0.01, seed Some(3L)))))) }3.1BatchEvaluation绑定引擎与评估器Evaluationtrait 的定义见 Evaluation.scala。它通过engineEvaluator这个 setter 接收「引擎 评估器」二元组内部会校验「评估器最多只能设置一次」。这里绑定的是RecommendationEngine()模板自带的引擎工厂见模板 Engine.scala内部把DataSource、Preparator、ALSAlgorithm、Serving组装在一起和新建的BatchPersistableEvaluator。对比默认模板常规的RecommendationEvaluation绑定的是MetricEvaluator(metric PrecisionAtK(...), otherMetrics ...)见 customize-serving 示例的 Evaluation.scala而这里换成了不做指标计算的BatchPersistableEvaluator。3.2BatchEngineParamsList指定引擎参数EngineParamsGenerator是一个包含engineParamsList的 traitpio eval的第二个参数就指向它。这里的engineParamsList只包含一个EngineParamsdataSourceParams记得把appName从INVALID_APP_NAME改成你自己的应用名即你导入事件数据时使用的 app 名evalParams None表示不启用 k-fold 切分因为我们不需要DataSourceEvalParams批量查询完全由覆写后的readEval()提供algorithmParamsListALS 算法的参数rank 10隐因子数量、numIterations 20迭代次数、lambda 0.01正则化系数、seed Some(3L)随机种子。这些参数可以直接沿用你训练时验证过的一组值。3.3 参数解析的源码依据EngineParams与WorkflowParams在 core 模块 与 WorkflowParams.scala 中定义。其中WorkflowParams还暴露了batch本次运行的批次标签、verbose日志级别、saveModel是否持久化模型、skipSanityCheck、stopAfterRead、stopAfterPrepare等参数意味着评估工作流本身也可以通过命令行开关做细粒度控制例如调试数据源时用--stop-after-read提前中止。engineParamsList是一个Seq常规调参场景下可以放多组参数做网格搜索如 EngineParamsList 中对rank、numIterations的组合遍历而批量预测场景通常只需要一组——这也正是BatchPersistableEvaluator里require(engineEvalDataSet.size 1)的前提。第 4 步构建并运行批量评估4.1 构建在模板根目录执行$ pio build构建成功后控制台应输出[INFO] [Console$] Your engine is ready for training.4.2 运行执行pio eval第一个参数是Evaluation对象全名第二个参数是EngineParamsGenerator对象全名$ pio eval org.template.recommendation.BatchEvaluation org.template.recommendation.BatchEngineParamsList4.3 预期输出运行成功后你应该看到类似下面的日志[INFO] [BatchPersistableEvaluator] Writing result to disk [INFO] [BatchPersistableEvaluator] Result can be found in batch_result [INFO] [CoreWorkflow$] Updating evaluation instance with result: org.template.recommendation.BatchPersistableEvaluatorResult2f886889 [INFO] [CoreWorkflow$] runEvaluation completed解读这四行日志前两行来自BatchPersistableEvaluator自身的logger.info表明落盘开始与完成后两行来自评估工作流入口 CoreWorkflow.runEvaluation先打印runEvaluation started随后把评估实例写入数据库并更新其状态EVALCOMPLETED最后打印runEvaluation completed。由于BatchPersistableEvaluatorResult没有覆写toOneLiner等方法日志中展示的是对象的默认字符串表示。4.4 查看结果在输出目录batch_result/下你可以找到批量查询及其预测结果。saveAsTextFile产生的文件内容大致为每条记录一行 JSON例如{query:{user:1,num:10},predictedResult:{itemScores:[{item:i123,score:4.82},...]}} {query:{user:3,num:15},predictedResult:{itemScores:[{item:i456,score:3.91},...]}} {query:{user:5,num:20},predictedResult:{itemScores:[{item:i789,score:3.05},...]}}每个 JSON 对象对应一个Row(query, predictedResult)query是你传入的批量查询predictedResult.itemScores是该用户按得分降序排列的推荐条目列表条目数量不超过查询中的num条目结构对应模板Engine.scala中的ItemScore(item: String, score: Double)。拿到这个文件后你可以用任意脚本如 Python/awk解析把它导入业务数据库或用于离线分析。运行链路从命令到结果文件的源码级复盘最后把整条链路在源码层面对齐方便你排查问题或做二次开发命令入口pio eval Eval ParamsList定位到CoreWorkflow.runEvaluationCoreWorkflow.scala它先创建评估实例记录再调用EvaluationWorkflow.runEvaluation(...)把evaluation、engine、engineParamsList、evaluator交给评估工作流数据准备评估工作流按EngineParams实例化引擎引擎内的 DataSource 调用你覆写后的readEval()产出批量Query及 dummyActualResult训练与预测引擎使用ALSAlgorithmParams(rank10, numIterations20, lambda0.01, seedSome(3))训练模型并对每个Query产出PredictedResult评估落盘评估工作流把RDD[(Query, PredictedResult, ActualResult)]交给BatchPersistableEvaluator.evaluateBase后者过滤掉实际结果、序列化Row(query, predictedResult)并saveAsTextFile(batch_result)收尾evaluateBase返回BatchPersistableEvaluatorResultCoreWorkflow.runEvaluation更新评估实例状态为EVALCOMPLETED并打印runEvaluation completedCoreWorkflow.scala。常见问题与注意事项appName未修改BatchEngineParamsList里默认是INVALID_APP_NAME不改成你的事件应用名会导致读取不到数据相对路径的输出目录outputDir batch_result是相对路径。在本地单机运行时没有问题在集群上运行如 Spark on YARN时任务可能在不同节点执行务必改用绝对路径或使用支持 Hadoop 接口的存储如 HDFS并通过绝对路径写入require约束如果修改了readEval()让它返回多个数据集或让engineParamsList包含多组参数BatchPersistableEvaluator会因require失败而中止——要么保持单数据集、单参数组要么按注释改造输出逻辑为不同参数组写不同目录特性稳定性BaseEvaluator属于DeveloperApi本文方案依赖的接口均为实验性 API升级 PredictionIO 版本后需要回归验证与常规评估的取舍如果你关心的是「哪组参数更好」请继续使用MetricEvaluator与PrecisionAtK参考 Evaluation.scala 示例如果你关心的是「给定一组参数为这批用户批量产出推荐结果」本文的BatchPersistableEvaluator正是为此设计。【免费下载链接】predictionioPredictionIO, a machine learning server for developers and ML engineers.项目地址: https://gitcode.com/gh_mirrors/pred/predictionio创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考