小编Sor*_*mar的帖子

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

它运行正常,但Dynamo中的记录被插入两次.

以下是我的代码.

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

如果我删除行以抛出异常,它工作正常但步骤已完成.

如何在不获取日志消息两次的情况下使步骤失败.

谢谢您的帮助.

此致,索拉布

我有2个数据框，我想查找除2（surrogate_key，当前）以外的所有列均相等的记录

然后，我想用新的surrogate_key值保存这些记录。

以下是我的代码：

val seq = csvDataFrame.columns.toSeq
var exceptDF = csvDataFrame.except(csvDataFrame.as('a).join(table.as('b),seq).drop("surrogate_key","current"))
exceptDF.show()

exceptDF = exceptDF.withColumn("surrogate_key", makeSurrogate(csvDataFrame("name"), lit("ecc")))
exceptDF = exceptDF.withColumn("current", lit("Y"))

exceptDF.show()

exceptDF.write.option("driver","org.postgresql.Driver").mode(SaveMode.Append).jdbc(postgreSQLProp.getProperty("url"), tableName, postgreSQLProp)

这段代码给出了正确的结果，但是在将这些结果写入postgre时却卡住了。

不确定是什么问题。还有没有更好的方法呢？

问候，Sorabh

我们可以使用以下方法从Java对象列表中创建一个数据框：

DataFrame df = sqlContext.createDataFrame(list, Example.class);

对于Java，Spark可以直接从类（在这种情况下）推断模式Example.class。

如果有Scala，有没有办法做同样的事情？