1

在 Java Spark 中，我有一个包含“bucket_timestamp”列的数据框，该列表示该行所属的存储桶的时间。

我想将数据帧写入 Cassandra DB。数据必须使用 TTL 写入 DB。TTL 应取决于存储桶时间戳 - 每行的 TTL 应计算为ROW_TTL = CONST_TTL - (CurrentTime - bucket_timestamp)，其中CONST_TTL是我配置的常量 TTL。

目前我正在使用恒定的 TTL 用 spark 写信给 Cassandra，代码如下：

df.write().format("org.apache.spark.sql.cassandra")
            .options(new HashMap<String, String>() {
                {
                    put("keyspace", "key_space_name");
                    put("table, "table_name");
                    put("spark.cassandra.output.ttl, Long.toString(CONST_TTL)); // Should be depended on bucket_timestamp column
                }
            }).mode(SaveMode.Overwrite).save();

我想到的一种可能的方法是 - 对于每个可能的 bucket_timestamp - 根据时间戳过滤数据，计算 TTL 并将过滤后的数据写入 Cassandra。但这似乎非常低效，而不是火花方式。Java Spark 中有没有办法提供一个 spark 列作为 TTL 选项，以便每一行的 TTL 都不同？

解决方案应该使用 Java 和数据集<行>：我遇到了一些在 scala 中使用 RDD 执行此操作的解决方案，但没有找到使用 Java 和数据框的解决方案。

谢谢！

4

2 回答 2

2

从 Spark-Cassandra 连接器选项（https://github.com/datastax/spark-cassandra-connector/blob/v2.3.0/spark-cassandra-connector/src/main/java/com/datastax/spark/connector/japi /RDDAndDStreamCommonJavaFunctions.java）您可以将 TTL 设置为：

  • 常数值 ( withConstantTTL)
  • 自动解析值 ( withAutoTTL)
  • 基于列的值 ( withPerRowTTL)

在您的情况下，您可以尝试最后一个选项并将 TTL 计算为以Dataset您在问题中提供的规则开头的新列。

对于用例，您可以在此处查看测试：https ://github.com/datastax/spark-cassandra-connector/blob/master/spark-cassandra-connector/src/it/scala/com/datastax/spark/connector/ writer/TableWriterSpec.scala#L612

于 2018-07-02T11:13:04.290 回答
1

对于 DataFrame API，尚不支持此类功能，但是...有 JIRA - https://datastax-oss.atlassian.net/browse/SPARKC-416，您可以观看它以在实现时收到通知。 ..

因此，您唯一的选择是使用 @bartosz25 的答案中描述的 RDD API...

于 2018-07-02T13:44:35.090 回答