Spark，在 EMR 中引发 SparkException 时行为不正确答案

【问题标题】：Spark, Incorrect behaviour when throwing SparkException in EMRSpark，在 EMR 中引发 SparkException 时行为不正确
【发布时间】：2018-03-07 06:28:23
【问题描述】：

我正在使用 YARN 作为资源管理器并在 2 个节点上在 EMR 中运行 spark 作业。如果不满足我的条件，我需要故意使该步骤失败，因此下一步不会按照配置执行。为了实现这一点，我在 dynamoDB 中插入日志消息后抛出了一个自定义异常。

它运行良好，但 Dynamo 中的记录被插入了两次。

下面是我的代码。

if(<condition>) {
  <method call to insert in dynamo> 
  throw new SparkException(<msg>);
  return;
}

如果我删除该行以引发异常，它可以正常工作，但该步骤已完成。

如何使步骤失败，而不会收到两次日志消息。

感谢您的帮助。

问候，索拉布

【问题讨论】：

标签： apache-spark amazon-dynamodb hadoop-yarn amazon-emr

【解决方案1】：

您的发电机消息被插入两次的原因可能是因为您的错误条件被两个不同的执行程序命中和处理。 Spark 正在将要完成的工作分配给它的工人，而这些工人不分享任何知识。

我不确定是什么促使您要求 Spark 步骤失败，但我建议您改为在应用程序代码中跟踪该失败案例，而不是尝试让 Spark 直接死掉。换句话说，编写检测错误并将其传递回火花驱动程序的代码，然后酌情采取行动。

一种方法是使用累加器来计算处理数据时发生的任何错误。它看起来大致像这样（我假设是 scala 和 DataFrames，但您可以根据需要适应 RDD 和/或 python）：

val accum = sc.longAccumulator("Error Counter")
def doProcessing(a: String, b: String): String = {
   if(condition) {
     accum.add(1)
     null
   }
   else {
     doComputation(a, b)
   }
}
val doProcessingUdf = udf(doProcessing _)

df = df.withColumn("result", doProcessing($"a", $"b"))

df.write.format(..).save(..)  // Accumulator value not computed until an action occurs!

if(accum.value > 0) {
    // An error detected during computation! Do whatever needs to be done.
    <insert dynamo message here>
}

这种方法的一个好处是，如果您在 Spark UI 中寻找反馈，您将能够在它运行时看到累加器的值。作为参考，这里是关于累加器的文档： http://spark.apache.org/docs/latest/rdd-programming-guide.html#accumulators

【讨论】：