如何在 spark 2.3.1 api 中使用累加器

Question

我正在使用带有 Cassandra 3.x 的 spark-sql_2.11-2.3.1 版本。 我需要提供一个具有

的验证功能

   column_family_name text,
    oracle_count bigint,
    cassandra_count bigint,
    create_timestamp timestamp,
    last_update_timestamp timestamp,
    update_user text

同样，我需要计算成功插入的记录数，即要填充的 cassandra_count，因为我想使用 spark 累加器。但不幸的是，我无法使用 spark-sql_2.11-2.3.1 版本找到所需的 API 示例。

以下是我保存到 cassandra sn-p 的内容

 o_model_df.write.format("org.apache.spark.sql.cassandra")
    .options(Map( "table" -> columnFamilyName, "keyspace" -> keyspace ))
    .mode(SaveMode.Append)
    .save()

这里如何为成功保存到 Cassandra 中的每一行实现累加器增量 ...

任何帮助将不胜感激。

Answer 1

Spark 的累加器通常用于您编写的转换中，不要指望 spark cassandra 连接器会为您提供类似的东西。

但总的来说 - 如果您的工作没有错误地完成，那么这意味着数据已正确写入数据库。

如果你想检查数据库中真正有多少行，那么你需要统计数据库中的数据——你可以使用 spark cassandra 连接器的cassandraCount 方法。主要原因 - 您的 DataFrame 中可能有多行可以映射到单个 Cassandra 行（例如，如果您错误地定义了主键，那么多行就有它）。