问题标签 [rocksdb-java]

你如何在rocksdb中以列格式写入数据。

列的示例可以是Id, state, URL, fieldNames, query;

我应该能够读取所有提供的数据Id。

我在使用 kstream 连接时遇到问题。我所做的是从一个主题中分离出 3 种不同类型的消息到新的流中。然后用两个创建另一个流的流进行一次内连接，最后我对新流和最后一个剩余流进行最后一次左连接。

连接的窗口时间为 30 秒。

这样做是为了过滤掉一些被其他人覆盖的消息。

我在 kubernetes 上运行此应用程序，并且 pod 的磁盘空间无限增长，直到 pod 崩溃。

我意识到这是因为连接将数据本地存储在 tmp/kafka-streams 目录中。

这些目录被称为：KSTREAM-JOINTHIS... KSTREAM-OUTEROTHER..

它存储来自 RocksDb 的 sst 文件，并且这些文件无限增长。

我的理解是，因为我使用 30 秒的窗口时间，所以这些应该在特定时间后被清除，但不是。

我还将 WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG 更改为 10 分钟，看看是否会发生变化，但事实并非如此。

我需要了解如何改变这种情况。

我们的 kafka 流应用程序会不断打开新的文件描述符，只要它们是新传入的消息，而不会关闭旧的。它最终导致异常。我们已将开放 fds 的限制提高到 65k，但似乎没有帮助。

Kafka 代理和 Kafka 流库都是 2.1 版

日志中不断出现的错误消息是：

org.apache.kafka.streams.processor.internals.StreamThread.run StreamThread.java：747 org.apache.kafka.streams.processor.internals.StreamThread.runLoop StreamThread.java：777 org.apache.kafka.streams.processor。 internals.StreamThread.runOnce StreamThread.java：883 org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit StreamThread.java：1029 org.apache.kafka.streams.processor.internals.TaskManager.commitAll TaskManager.java：405 org.apache.kafka.streams.processor.internals.AssignedTasks.commit AssignedTasks.java：346 org.apache.kafka.streams.processor.internals.StreamTask.commit StreamTask.java：431 org.apache.kafka.streams.processor。 internals.StreamTask.commit StreamTask.java：443 org.apache.kafka.streams.processor.internals.StreamTask.flushState StreamTask.java：491 org.apache.kafka.streams。processor.internals.AbstractTask.flushState AbstractTask.java: 204 org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush ProcessorStateManager.java: 217 org.apache.kafka.streams.state.internals.MeteredKeyValueStore.flush MeteredKeyValueStore.java ：226 org.apache.kafka.streams.state.internals.WrappedStateStore$AbstractStateStore.flush WrappedStateStore.java：85 org.apache.kafka.streams.state.internals.RocksDBStore.flush RocksDBStore.java：388 org.apache.kafka。 stream.state.internals.RocksDBStore.flushInternal RocksDBStore.java: 395 org.rocksdb.RocksDB.flush RocksDB.java: 1743 org.rocksdb.RocksDB.flush RocksDB.java org.rocksdb.RocksDBException: 打开文件追加时：/ tmp/kafka-streams/s4l-notifications-test/5_1/rocksdb/main-store/002052.sst：打开文件过多状态：#object [org.rocksdb。状态 0x1cca4c5c "org.rocksdb.Status@1cca4c5c"] org.apache.kafka.streams.errors.ProcessorStateException：从存储主存储执行刷新时出错

任何想法如何调试它？

我对rockdb非常陌生，将在我的应用程序中使用rockdb作为查找服务。提供的rockDBjava库api是否包含数据库本身，我的意思是不需要单独安装rockDB数据库。

我尝试使用库运行代码并查看在代码中提到的 db 路径中创建的一些文件，因此不确定它是如何工作的，以及是否需要单独安装 DB 或者库是否将数据存储在目录中并且它本身充当数据库？

我需要聚合一些传感器数据，并且我正在为 PoC/研究项目尝试窗口聚合。

在到处研究并进行了很多尝试之后，我想出了以下代码，这似乎可以正常工作。事实上，我得到了显示预期结果的窥视，我还可以看到 SensorData 对象的反序列化/序列化通过流发生。

我可以在输出主题中看到最终结果。

但问题是：在输出主题中写入第一个结果后，流会崩溃，并且它会根据 commitMs 的长度提前或延迟崩溃。

任何想法？我在这里做错了吗？我正在使用landoop/fast-data-dev docker 镜像来测试它。

这是堆栈跟踪的相关部分：

我有rocksDB java实现，我想只在内存中存储数据而不把它写在硬盘上。我想我可以在创建 RocksDB 时使用选项来做到这一点。我试图使这里写的内容相同：https ://github.com/facebook/rocksdb/wiki/RocksDB-Tuning-Guide在“具有完整功能的内存数据库”部分中，但我不能完全相同，因为它是关于c++ 实现和 Java 要么没有某些属性，要么它们不可编辑。

我正在设置一个新的 Kafka 流应用程序，并希望使用 RocksDb 使用自定义状态存储。这适用于将数据放入状态存储并从中获取可查询的状态存储并迭代数据，但是，大约 72 小时后，我观察到存储中缺少数据。Kafka 流或 RocksDb 中状态存储的数据是否有默认保留时间？

我正在使用使用 RocksDb 的自定义状态存储，以便我们可以利用列族功能，我们不能将其与带有 KStreams 的嵌入式 RocksDb 实现一起使用。我已经使用 KeyValueStore 接口实现了自定义存储。还有我自己的 StoreSupplier、StoreBuilder、StoreType 和 StoreWrapper。为应用程序创建了一个变更日志主题，但还没有数据进入它（还没有研究这个问题）。

将数据放入此自定义状态存储并从中获取可查询的状态存储工作正常。但是，我发现在距商店约 72 小时后丢失了数据。我通过获取状态存储目录的大小以及将数据导出到文件并检查条目数来进行检查。

使用 SNAPPY 压缩和 UNIVERSAL 压缩

简单拓扑：

来自自定义商店实现的片段：

期望状态存储（RocksDb）将无限期地保留数据，直到手动删除或存储磁盘出现故障。我不知道 Kafka 流已经引入了带有状态存储的 TTl。

有一个用例，我必须读取巨大的 Parquet 文件并转换为 Rocksdb 二进制文件，所以我决定使用 spark（因为我团队中的每个人都熟悉它）。

从 Rocksdb 方面我知道它不是分布式的，你不能并行化。

所以我所做的是我为每个任务使用 Spark 并行创建了多个 Rocksdb 实例。

现在我想将它们结合在一起。所以我的问题是，是否可以将 Rocksdb 的多个实例组合在一起以使用一些后处理来创建一个大的 Rocksdb 实例？

我是 RocksDB abd 的新手，试图在 Java 中创建一个 SST 文件以进行批量加载。最终用例是在 Apache Spark 中创建它。

我rocksdbjni在 Ubuntu 18.04.03 中使用 6.3.6

我不断收到此错误，

示例代码是

如果循环索引小于 10，则文件创建成功，我能够将其摄取到 Rocks db 中。

提前致谢。

我正在使用 RocksDB JNI，我发现该程序的读取速度呈指数级增长

最初，它在可接受的范围内。但是当程序运行以使总记录达到 100 万条时，读取时间记录器打印显示大约 200-300 毫秒。随着程序的运行，情况变得更糟。我是否使用错误的 JNI？