你需要
坚持训练模型（例如。 modelFitted )你的流媒体工作可以访问的地方。这是在你的流媒体工作之外完成的。

modelFitted.write.overwrite().save("/path/to/model/location")

然后在结构化流式处理作业中加载此模型

import org.apache.spark.ml._
val model = PipelineModel.read.load("/path/to/model/location")

将此模型应用于流Dataframe（例如。 df )与

model.transform(df)

// in your case you may work with two streaming Dataframes to apply `approxSimilarityJoin`.

可能需要将流Dataframe转换成正确的格式，以便在模型预测中使用。