Spark：加入待转换的数据集时，“SparkException：Task not serializable”

Posted 2021-04-19

我有以下代码。它在两个Datasets上执行连接操作，其中一个在join转换中被过滤。

activeUserProfileDataset.join(
      allJobModelsDataset.filter(jobModel => jobIdRecCandidatesBroadcasted.value.contains(jobModel.JobId)),
      $"notVisitedJobId" === col(JobModelFieldNames.jobId),
      "left_outer")

这导致了问题：

SparkException：任务不可序列化

然而，当我拿出filter变换并在Dataset之外创建第二个join时，这个时间有效：

val jobIdRecCandidatesJobModels = allJobModelDataset.filter(jobModel => jobIdRecCandidatesBroadcasted.value.contains(jobModel.JobId))

val userJobPredictionsDataset3 = userJobPredictionsDataset2.join(
      jobIdRecCandidatesJobModels,
      $"notVisitedJobId" === col(JobModelFieldNames.jobId),
      "left_outer") 

你能解释一下这是为什么吗？你能告诉我这些转换操作（如join，filter）是如何在内部工作的吗？谢谢！

答案

这是因为在Spark中，您无法在其他转换中指定转换。主要思想：

• 驱动程序节点处理DAG并创建任务
• 工人执行转换（以任务的形式）

在第一个示例中，您尝试处理DAG并在转换内创建任务（在工作节点上）。通常，您需要创建一个需要在另一个DF上创建任务的任务。但请记住 - 工人无法创造新任务。他们只执行它们。

在第二个示例中 - 您在驱动程序节点上正确执行所有操例如。首先在DF上创建转换，然后在新任务中使用生成的DF。

希望有帮助:)