问题标签 [rocksdb]

我的启动方法：

我的关机方法：

但是关闭完成后，我仍然收到mergestack underrocksdb::DBImpl::BGWorkCompaction(void * arg)的请求，这当然会失败，因为所有列族句柄都已被处理掉。

如何标记任何压实或冲洗以停止？删除数据库实例似乎还不够。

我的迭代器代码：

迭代的顺序无关紧要，但打破循环很重要，所以我们不会迭代不需要的元素。我使用元素按有意义的顺序排序的事实来知道何时中断。callback可能需要很长时间。

所以我想做的伪代码是

我尝试应用 concurrency::parallel_for_each，但似乎不适合rocksdb api。

你会如何建议实现并发迭代？

我无法理解 Spark 如何与存储交互。

我想创建一个从 RocksDB 数据库（或任何其他键值存储）中获取数据的 Spark 集群。但是，此时，我能做的最好的事情就是将整个数据集从数据库中提取到每个集群节点的内存中（例如，放入地图中），然后从该对象构建一个 RDD。

我必须做什么才能只获取必要的数据（就像 Spark 对 HDFS 所做的那样）？我已经阅读了有关 Hadoop 输入格式和记录读取器的信息，但我并没有完全掌握我应该实现的内容。

我知道这是一个广泛的问题，但我非常感谢一些帮助让我开始。先感谢您。

我有几台机器处理在 RocksDB 中索引的大量文本数据（100 GB）。这些机器用于负载平衡并在相同的数据上运行。添加新机器时，我想尽快通过网络从现有机器复制数据库。

有没有一种优雅的方法可以通过网络进行 RocksDB 备份？我已阅读https://github.com/facebook/rocksdb/wiki/How-to-backup-RocksDB但这需要两倍的磁盘空间：首先要备份到本地文件系统，然后才能将其复制过来网络。我还必须处理例如 rsyncing 文件。

有谁知道如何用 snappy 构建 shared_lib ？

在过去的 3 个小时里，我尝试了：

1. 设置环境变量WITH_SNAPPY
2. 编辑thirdparty.inc并更改USE_SNAPPY_DEFAULT为 1
3. 传递变量以通过make shared_lib EXTRA_CFLAGS=-DSNAPPY
4. 与手鼓共舞。

没有什么帮助。我应该怎么办 ？

CompositeKey我有一个名为<userId: int, attribute: string>. 另一个名为SimpleValue.

使用 Kafka Streams，我使用 Avro 序列化器/反序列化器将它们放入 KeyValueStore。

稍后，使用交互式查询我可以get userId, attribute一对一配对。但我也想得到所有attributes属于一个单一的userId. 当我使用时，

它不会返回所有属性。我猜这是因为底层的 RocksDB 像字节数组一样对它们进行排序，而 avro 序列化格式与此混淆。

有没有办法在 RocksDB 中使用 Avro 进行部分范围查询？否则，我将不得不将它们与从 Avro 模式生成的字符串一起以"<userId>+<attribute>".

为什么“Ref”如此特别？

有谁能够帮我？

我正在使用 Kafka Streams 0.10.1.1 版本。

状态存储的 RocksDB 实现无法处理我们的 50k/msg 速率，所以我想将状态存储更改为内存中的存储。根据文档，这应该是可能的：http: //docs.confluent.io/3.1.0/streams/architecture.html#state

但是，当我实现这个时：

我最终在运行时出现此错误：

上述方法的实现是：

为什么它试图将状态存储转换为CachedStateStore实例？我怎样才能实现一个简单的内存状态存储，根据文档应该是可能的？

谢谢

根据生存时间文档：

在 t=0 时打开 1，ttl=4 并插入 k1，k2，在 t=2 时关闭 t=3 时打开2，ttl=5。现在 k1,k2 应该在 t>=5 时被删除

在我的用例中：

1. 如果我在 t = 0 时使用 ttl = 4 打开 Rocksdb
2. 插入 k1，k2。
3. 在 t = 2 时关闭数据库。
4. ttl=5 在 t=4 再次打开 DB
5. 插入 k1，k3。

现在 k1、k2 和 k3 的 ttl 是多少？ 记录的 TTL 是否取决于多次打开数据库？

我在想所有记录（k1，k2，k3）的 ttl 会被更改，因为我们已经多次打开 DB，即使使用相同的 TTL。

我创建了一个 kafka 流应用程序。在那，我试图将 KStream 聚合到 KTable 中。在那里，我指定了一个供 Kafka 内部使用的 statestore。但是，在运行应用程序时，我遇到了错误。

卡夫卡版本：10.2.0