rocksdb使用_flink写入kafka

小小码 • 2023-01-26 09:00 • 代码基础 • 阅读 167

大家好，我是考100分的小小码，祝大家学习进步，加薪顺利呀。今天说一说rocksdb使用_flink写入kafka,希望您对编程的造诣更进一步.

「Flink」RocksDB介绍以及Flink对RocksDB的支持

RocksDB介绍

RocksDB简介

RocksDB是基于C++语言编写的嵌入式KV存储引擎，它不是一个分布式的DB，而是一个高效、高性能、单点的数据库引擎。它是由Facebook基于Google开源的kv存储LevelDB开发开发。RocksDB使用LSM存储引擎。它针对不同的生产环境进行调优，可以直接使用内存、也可以使用Flash、或者用硬盘或者HDFS。而且支持不同的压缩算法，有一整套的工具用于生产、调试使用。RocksDB是一种嵌入式、KV型、持久化的存储。

使用嵌入式的数据存储原因有很多，当数据频繁访问内存、或者存储时，网络延迟会增加响应时间。

RocksDB的主要应用场景

适应于多CPU场景

一般的商业服务器有很多的CPU核，例如：志强E5系列 – 6核
RocksDB可以高效运行在多核服务器上
它提供的RocksDB语义比传统DBMS更简单

高效利用存储

RocksDB可以在快速存储上高效运行且不会成为性能瓶颈
RocksDB采用LSM引擎，对比B-Tree引擎，它有更好的压缩和更小的写放大

弹性架构，支持扩展
支持IO-bound、in-memory、write-once

入门案例

为了简单说明RocksDB，我们这里使用RocksDB的Java版本来编写。

导入Maven依赖

<dependencies>
        <!-- https://mvnrepository.com/artifact/org.rocksdb/rocksdbjni -->
        <dependency>
            <groupId>org.rocksdb</groupId>
            <artifactId>rocksdbjni</artifactId>
            <version>5.11.3</version>
        </dependency>
    </dependencies>

代码100分

基于RocksDB读写数据

代码100分public class GettingStartDemo {
    // 因为RocksDB是由C++编写的，在Java中使用首先需要加载Native库
    static {
        // Loads the necessary library files.
        // Calling this method twice will have no effect.
        // By default the method extracts the shared library for loading at
        // java.io.tmpdir, however, you can override this temporary location by
        // setting the environment variable ROCKSDB_SHAREDLIB_DIR.
        // 默认这个方法会加压一个共享库到java.io.tmpdir
        RocksDB.loadLibrary();
    }

    public static void main(String[] args) throws RocksDBException {
        // 1. 打开数据库
        // 1.1 创建数据库配置
        Options dbOpt = new Options();
        // 1.2 配置当数据库不存在时自动创建
        dbOpt.setCreateIfMissing(true);
        // 1.3 打开数据库。因为RocksDB默认是保存在本地磁盘，所以需要指定位置
        RocksDB rdb = RocksDB.open(dbOpt, "./data/rocksdb");

        // 2. 写入数据
        // 2.1 RocksDB都是以字节流的方式写入数据库中，所以我们需要将字符串转换为字节流再写入。这点类似于HBase
        byte[] key = "zhangsan".getBytes();
        byte[] value = "20".getBytes();

        // 2.2 调用put方法写入数据
        rdb.put(key, value);
        System.out.println("写入数据到RocksDB完成！");

        // 3. 调用delete方法读取数据
        System.out.println("从RocksDB读取key = " + new String(key) + "的value为" + new String(rdb.get(key)));

        // 4. 移除数据
        rdb.delete(key);

        // 关闭资源
        rdb.close();
        dbOpt.close();
    }
}

运行程序后，我们可以发现，在data/rocksdb文件夹中，生成了一下几个文件：

0000004.sst

sst是RocksDB的数据存储文件，是二进制格式的

0000006.log

log是预写日志文件，LSM架构引擎都是有预写日志的

CURRENT

CURRENT文件是一个文本文件，记录最近的MANIFEST

IDENTITY

存放当前rocksdb的唯一标识

LOCK

LOCK 进程的全局锁，DB一旦被open, 其他进程将无法修改

rocksdb的操作日志文件，可配置定期的统计信息写入LOG. 可通过info_log_level调整日志输出级别; 通过keep_log_file_num限制文件数量等等。

LOG.old.15807….
MANIFECT-000005

记录rocksdb最近的状态变化日志。其中包含manifest日志和最新的文件指针

OPTIONS-000005

rocksdb的配置文件

OPTIONS-000008

Flink使用RocksDBBackend

1、导入Maven依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
    <version>1.9.0</version>
</dependency>

2、配置启用RocksDBBackend

代码100分// 此处也可以是HDFS路径，这里为了测试方便，所以使用的是本地路径

env.setStateBackend(new RocksDBStateBackend("file:///D:/project/java8/data/rocksdb", true));

参考文献：

RocksDB中文网：https://rocksdb.org.cn/

https://rocksdb.org.cn/doc/RocksJava-Basics.html

https://www.jianshu.com/p/2638e2b379c3

https://www.jianshu.com/p/3302be5542c7

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至举报，一经查实，本站将立刻删除。
转载请注明出处: https://daima100.com/10088.html