是否可以在Spark结构化流中使用foreachBatch将两个不相交的数据集写入数据同步中？

问题描述

我正在尝试将数据从单一来源写入多个DataSink（Mongo和Postgres DB）。 传入数据

Dataset<Row> df = spark
        .readStream()
        .format("kafka")
        .option("kafka.bootstrap.servers","localhost:9092")
        .option("subscribe","topic1")
        .load();

Dataset<Row> personalDetails = df.selectExpr("name","id","age");


personalDetails.writeStream()
    .outputMode(OutputMode.Update())
    .foreachBatch((dataframe,bachId) -> {
            dataframe.write().format("com.mongodb.spark.sql.DefaultSource").mode(SaveMode.Append)
                    .option("uri","mongodb://localhost/employee")
                    .option("database","employee")
                    .option("collection","PI").save();
    }).start();

Dataset<Row> salDetails = df.selectExpr("basicSal","bonus");
salDetails.writeStream()
    .outputMode(OutputMode.Update())
    .foreachBatch((dataframe,"SAL").save();
    }).start();

问题是，我可以看到Spark正在打开两个Stream，并且两次读取相同的事件。 是否可以读取一次并应用不同的转换并写入不同的集合？

解决方法

您应该缓存DataFrame。 参见here：

写入多个位置-如果要将流查询的输出写入多个位置，则只需将输出DataFrame / Dataset多次写入即可。但是，每次写入尝试都可能导致重新计算输出数据（包括可能重新读取输入数据）。为了避免重新计算，您应该缓存输出DataFrame / Dataset，将其写入多个位置，然后取消缓存。

及其示例：

streamingDF.writeStream.foreachBatch { (batchDF: DataFrame,batchId: Long) =>
  batchDF.persist()
  batchDF.write.format(...).save(...)  // location 1
  batchDF.write.format(...).save(...)  // location 2
  batchDF.unpersist()
}

您可以将所有代码放在一个foreachBatch中，并将数据帧写入两个接收器。您可以通过缓存数据帧，然后在此缓存的数据帧上执行selectExpr并保存它来完成此操作。

作为旁注-请注意，无论如何，如果您想要“全有或全无”（即您不希望自己写信给mongo而不是写postgres的情况），您只能使用一个foreachBatch，因为否则（如果您有2个foreachBatch，例如您的问题），则有2个独立的批次-对于相同的数据，一个批次可能会失败，而另一个批次可能会失败。