• 构建高效实时数据流水线：Flink、Kafka 和 CnosDB 的完美组合

    当今的数据技术生态系统中，实时数据处理已经成为许多企业不可或缺的一部分。为了满足这种需求，Apache Flink、Apache Kafka和CnosDB等开源工具的结合应运而生，使得实时数据流的收集、处理和存储变得更加高效和可靠。本篇文章将介绍如何使用 Flink、Kafka 和 CnosDB 来构建一个强大的实时数据处理流水线。

    什么是 Flink、Kafka、CnosDB

    • Flink：是一个强大的流式处理引擎，它支持事件驱动、分布式、并且容错。Flink能够处理高吞吐量和低延迟的实时数据流，适用于多种应用场景，如数据分析、实时报表和推荐系统等。
    • Kafka：是一个高吞吐量的分布式流数据平台，用于收集、存储和传输实时数据流。Kafka具有良好的持久性、可扩展性和容错性，适用于构建实时数据流的可靠管道。
    • CnosDB：是一个专为时序数据设计的开源时序数据库。它具有高性能、高可用性和易用性的特性，非常适合存储实时生成的时间序列数据，如传感器数据、日志和监控数据等。

    场景描述

    用例中假设有一个物联网设备网络，每个设备都定期生成传感器数据，包括温度、湿度和压力等。我们希望能够实时地收集、处理和存储这些数据，以便进行实时监控和分析。

    数据流向架构图如下：

    1. 首先，我们需要设置一个数据收集器来获取传感器数据，并将数据发送到 Kafka 主题。这可以通过编写一个生产者应用程序来实现，该应用程序将生成的传感器数据发送到 Kafka。
    2. 使用 Flink来实时处理传感器数据。首先，需要编写一个Flink应用程序，该应用程序订阅 Kafka 主题中的数据流，并对数据进行实时处理和转换。例如，您可以计算温度的平均值、湿度的最大值等。
    3. 将处理后的数据存储到 CnosDB 中以供后续查询。为了实现这一步，需要配置一个CnosDB Sink，使得Flink应用程序可以将处理后的数据写入 CnosDB 中。

    构建流水线

    1.数据采集与传输

    编写一个生产者应用程序，读取传感器数据并将其发送到 Kafka 主题。

    1. public class SensorDataProducer {
    2.     public static void main(String[] args) {
    3.         Properties props = new Properties();
    4.         props.put("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092");
    5.         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    6.         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    7.  
    8.  
    9.         KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    10.  
    11.  
    12.         while (true) {
    13.             SensorData data = generateSensorData(); // 生成传感器数据
    14.             producer.send(new ProducerRecord<>("sensor-data-topic", data));
    15.             Thread.sleep(1000); // 每秒发送一次数据
    16.         }
    17.     }
    18. }

    2.实时处理与转换

    编写一个 Flink 应用程序，订阅 Kafka 主题中的数据流，实时处理并转换数据。

    1. // Flink 应用程序示例
    2. public class SensorDataProcessingJob {
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.  
    6.  
    7.         Properties props = new Properties();
    8.         props.setProperty("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092");
    9.         props.setProperty("group.id", "sensor-data-consumer-group");
    10.  
    11.  
    12.         DataStream<String> sensorData = env.addSource(new FlinkKafkaConsumer<>("sensor-data-topic", new SimpleStringSchema(), props));
    13.  
    14.  
    15.         DataStream<ProcessedData> processedData = sensorData
    16.             .map(json -> parseJson(json)) // 解析JSON数据
    17.             .keyBy(ProcessedData::getDeviceId)
    18.             .window(TumblingEventTimeWindows.of(Time.seconds(10))) // 10秒滚动窗口
    19.             .apply(new SensorDataProcessor()); // 自定义处理逻辑
    20.  
    21.  
    22.         processedData.print(); // 打印处理后的数据，可以替换为写入 CnosDB 操作
    23.  
    24.  
    25.         env.execute("SensorDataProcessingJob");
    26.     }
    27. }

    3.数据写入与存储

    配置CnosDB Sink，将 processedData.print() 替换为写入 CnosDB 的程序在 CnosDB 创建一个存储数据时长为 30 天的数据库：

    | CnosDB 建库语法说明请查看：创建数据库[https://docs.cnosdb.com/zh/latest/reference/sql.html#创建数据库]

    CREATE DATABASE IF NOT EXISTS "db_flink_test" WITH TTL '30d' SHARD 2 VNODE_DURATION '1d' REPLICA 2;

    在 Maven [https://maven.apache.org/]中引入 CnosBD Sink [https://docs.cnosdb.com/zh/latest/reference/connector/flink-connector-cnosdb.html]包：

    1. <dependency>
    2.     <groupId>com.cnosdb</groupId>
    3.     <artifactId>flink-connector-cnosdb</artifactId>
    4.     <version>1.0</version>
    5. </dependency>

    编写程序：

    1. public class WriteToCnosDBJob {
    2.     public static void main(String[] args) throws Exception {
    3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    4.  
    5.  
    6.         Properties props = new Properties();
    7.         props.setProperty("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092");
    8.         props.setProperty("group.id", "sensor-data-consumer-group");
    9.  
    10.  
    11.         DataStream<String> sensorData = env.addSource(new FlinkKafkaConsumer<>("sensor-data-topic", new SimpleStringSchema(), props));
    12.  
    13.  
    14.         DataStream<ProcessedData> processedData = sensorData
    15.             .map((MapFunction<String, ProcessedData>) json -> parseJson(json)) // 解析JSON数据
    16.             .keyBy(ProcessedData::getDeviceId)
    17.             .window(TumblingEventTimeWindows.of(Time.seconds(10))) // 10秒滚动窗口
    18.             .apply(new SensorDataProcessor()); // 自定义处理逻辑
    19.  
    20.  
    21.         DataStream<CnosDBPoint> cnosDBDataStream = processedData.map(
    22.                 new RichMapFunction<ProcessedData, CnosDBPoint>() {
    23.                     @Override
    24.                     public CnosDBPoint map(String s) throws Exception {
    25.                         return new CnosDBPoint("sensor_metric")
    26.                                 .time(value.getTimestamp().toEpochMilli(), TimeUnit.MILLISECONDS)
    27.                                 .tag("device_id", value.getDeviceId())
    28.                                 .field("average_temperature", value.getAverageTemperature())
    29.                                 .field("max_humidity", value.getMaxHumidity());
    30.                     }
    31.                 }
    32.         );
    33.  
    34.  
    35.         CnosDBConfig cnosDBConfig = CnosDBConfig.builder()
    36.                 .url("http://localhost:8902")
    37.                 .database("db_flink_test")
    38.                 .username("root")
    39.                 .password("")
    40.                 .build();
    41.  
    42.  
    43.         cnosDBDataStream.addSink(new CnosDBSink(cnosDBConfig));
    44.         env.execute("WriteToCnosDBJob");
    45.     }
    46. }

    运行后查看结果：

    1. db_flink_test ❯ select * from sensor_metric limit 10;
    2. +---------------------+---------------+---------------------+--------------+
    3. | time                | device_id     | average_temperature | max_humidity |
    4. +---------------------+---------------+---------------------+--------------+
    5. | 2023-01-14T17:00:00 | OceanSensor1  | 23.5                | 79.0         |
    6. | 2023-01-14T17:05:00 | OceanSensor2  | 21.8                | 68.0         |
    7. | 2023-01-14T17:10:00 | OceanSensor1  | 25.2                | 75.0         |
    8. | 2023-01-14T17:15:00 | OceanSensor3  | 24.1                | 82.0         |
    9. | 2023-01-14T17:20:00 | OceanSensor2  | 22.7                | 71.0         |
    10. | 2023-01-14T17:25:00 | OceanSensor1  | 24.8                | 78.0         |
    11. | 2023-01-14T17:30:00 | OceanSensor3  | 23.6                | 80.0         |
    12. | 2023-01-14T17:35:00 | OceanSensor4  | 22.3                | 67.0         |
    13. | 2023-01-14T17:40:00 | OceanSensor2  | 25.9                | 76.0         |
    14. | 2023-01-14T17:45:00 | OceanSensor4  | 23.4                | 70.0         |
    15. +---------------------+---------------+---------------------+--------------+

    总结

    通过结合Flink、Kafka 和 CnosDB，您可以构建一个强大的实时数据处理流水线，从数据采集到实时处理再到数据存储和可视化。每个步骤都涉及具体的配置和代码实现，确保您熟悉每个工具的特性和操作。这种架构适用于各种实时数据应用，如物联网监控、实时报表和仪表板等。根据您的需求和情境，调整配置和代码，以构建适合您业务的实时数据处理解决方案。

    CnosDB简介

    CnosDB是一款高性能、高易用性的开源分布式时序数据库，现已正式发布及全部开源。

    欢迎关注我们的社区网站：https://cn.cnosdb.com

  • 相关阅读:
    nnDetection复现Luna16 附模型
    前端基础学习——你必须掌握的CSS的知识
    LVGL_基础控件checkbox
    Java：SynchronousQueue同步队列
    MyBatis源码剖析之Mapper代理方式细节
    专业135总分400+西安交通大学信息与通信工程学院909/815考研经验分享
    使用免费开源软件 Blender 编辑视频
    【喜报】云贝学员顺利通过OceanBase 数据库上机实验，OBCP证书到手了！！！
    python+nodejs+vue教材征订管理系统
    go 学习 之 HTTP微服务示例
  • 原文地址：https://blog.csdn.net/CnosDB/article/details/132656963