二百五十九、Java——采集Kafka数据，解析成一条条数据，写入另一Kafka中（一般JSON）

2024-09-04 07:04
文章标签 java 数据 json 解析 采集 写入 kafka 一般 二百五十 一条条

本文主要是介绍二百五十九、Java——采集Kafka数据，解析成一条条数据，写入另一Kafka中（一般JSON），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、目的

由于部分数据类型频率为1s，从而数据规模特别大，因此完整的JSON放在Hive中解析起来，尤其是在单机环境下，效率特别慢，无法满足业务需求。

而Flume的拦截器并不能很好的转换数据，因为只能采用Java方式，从Kafka的主题A中采集数据，并解析字段，然后写入到放在Kafka主题B中

二 、原始数据格式

JSON格式比较正常，对象中包含数组

{
    "deviceNo": "39",
    "sourceDeviceType": null,
    "sn": null,
    "model": null,
    "createTime": "2024-09-03 14:10:00",
    "data": {
        "cycle": 300,
        "evaluationList": [{
            "laneNo": 1,
            "laneType": null,
            "volume": 3,
            "queueLenMax": 11.43,
            "sampleNum": 0,
            "stopAvg": 0.54,
            "delayAvg": 0.0,
            "passRate": 0.0,
            "travelDist": 140.0,
            "travelTimeAvg": 0.0
        },
        {
            "laneNo": 2,
            "laneType": null,
            "volume": 7,
            "queueLenMax": 23.18,
            "sampleNum": 0,
            "stopAvg": 0.47,
            "delayAvg": 10.57,
            "passRate": 0.0,
            "travelDist": 140.0,
            "travelTimeAvg": 0.0
        },
        {
            "laneNo": 3,
            "laneType": null,
            "volume": 9,
            "queueLenMax": 11.54,
            "sampleNum": 0,
            "stopAvg": 0.18,
            "delayAvg": 9.67,
            "passRate": 0.0,
            "travelDist": 140.0,
            "travelTimeAvg": 0.0
        },
        {
            "laneNo": 4,
            "laneType": null,
            "volume": 6,
            "queueLenMax": 11.36,
            "sampleNum": 0,
            "stopAvg": 0.27,
            "delayAvg": 6.83,
            "passRate": 0.0,
            "travelDist": 140.0,
            "travelTimeAvg": 0.0
        }]
    }
}

三、Java代码

package com.kgc;import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class KafkaKafkaEvaluation {// 添加 Kafka Producer 配置private static Properties producerProps() {Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.70:9092");props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.ACKS_CONFIG, "-1");props.put(ProducerConfig.RETRIES_CONFIG, "3");props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");props.put(ProducerConfig.LINGER_MS_CONFIG, "1");props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");return props;}public static void main(String[] args) {Properties prop = new Properties();prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.70:9092");prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");prop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");// 每一个消费，都要定义不同的Group_IDprop.put(ConsumerConfig.GROUP_ID_CONFIG, "evaluation_group");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);consumer.subscribe(Collections.singleton("topic_internal_data_evaluation"));ObjectMapper mapper = new ObjectMapper();// 初始化 Kafka ProducerKafkaProducer<String, String> producer = new KafkaProducer<>(producerProps());while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {try {JsonNode rootNode = mapper.readTree(record.value());System.out.println("原始数据"+rootNode);String device_no = rootNode.get("deviceNo").asText();String source_device_type = rootNode.get("sourceDeviceType").asText();String sn = rootNode.get("sn").asText();String model = rootNode.get("model").asText();String create_time = rootNode.get("createTime").asText();String cycle = rootNode.get("data").get("cycle").asText();JsonNode evaluationList = rootNode.get("data").get("evaluationList");for (JsonNode evaluationItem : evaluationList) {String lane_no = evaluationItem.get("laneNo").asText();String lane_type = evaluationItem.get("laneType").asText();String volume = evaluationItem.get("volume").asText();String queue_len_max = evaluationItem.get("queueLenMax").asText();String sample_num = evaluationItem.get("sampleNum").asText();String stop_avg = evaluationItem.get("stopAvg").asText();String delay_avg = evaluationItem.get("delayAvg").asText();String pass_rate = evaluationItem.get("passRate").asText();String travel_dist = evaluationItem.get("travelDist").asText();String travel_time_avg = evaluationItem.get("travelTimeAvg").asText();String outputLine = String.format("%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s",device_no, source_device_type, sn, model, create_time, cycle,lane_no, lane_type,volume,queue_len_max,sample_num,stop_avg,delay_avg,pass_rate,travel_dist,travel_time_avg);// 发送数据到 KafkaProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic_db_data_evaluation", record.key(), outputLine);producer.send(producerRecord, (RecordMetadata metadata, Exception e) -> {if (e != null) {e.printStackTrace();} else {System.out.println("The offset of the record we just sent is: " + metadata.offset());}});}} catch (Exception e) {e.printStackTrace();}}consumer.commitAsync();}}}

1、服务器IP都是   192.168.0.70

2、消费Kafka主题（数据源）：topic_internal_data_evaluation

3、生产Kafka主题（目标源）：topic_db_data_evaluation

4、注意：字段顺序与ODS层表结构字段顺序一致！！！

四、开启Kafka主题topic_db_data_evaluation消费者

[root@localhost bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.0.70:9092  --topic topic_db_data_evaluation  --from-beginning

五、运行测试

1、启动项目

2、消费者输出数据

然后再用Flume采集写入HDFS就行了，不过ODS层表结构需要转变

六、ODS层新表结构

create external table  if not exists  hurys_dc_ods.ods_evaluation(device_no           string        COMMENT '设备编号',source_device_type  string        COMMENT '设备类型',sn                  string        COMMENT '设备序列号 ',model               string        COMMENT '设备型号',create_time         timestamp     COMMENT '创建时间',cycle               int           COMMENT '评价数据周期',lane_no             int           COMMENT '车道编号',lane_type           int           COMMENT '车道类型 0:渠化1:来向2:出口3:去向4:左弯待转区5:直行待行区6:右转专用道99:未定义车道',volume              int           COMMENT '车道内过停止线流量（辆）',queue_len_max       float         COMMENT '车道内最大排队长度（m）',sample_num          int           COMMENT '评价数据计算样本量',stop_avg            float         COMMENT '车道内平均停车次数（次）',delay_avg           float         COMMENT '车道内平均延误时间（s）',pass_rate           float         COMMENT '车道内一次通过率',travel_dist         float         COMMENT '车道内检测行程距离（m）',travel_time_avg     float         COMMENT '车道内平均行程时间'
)
comment '评价数据外部表——静态分区'
partitioned by (day string)
row format delimited fields terminated by ','
stored as SequenceFile
;

七、Flume采集配置文件

八、运行Flume任务，检查HDFS文件、以及ODS表数据

--刷新表分区
msck repair table ods_evaluation;
--查看表分区
show partitions hurys_dc_ods.ods_evaluation;
--查看表数据
select * from hurys_dc_ods.ods_evaluation
where day='2024-09-03';

搞定，这样就不需要在Hive中解析JSON数据了！！！

这篇关于二百五十九、Java——采集Kafka数据，解析成一条条数据，写入另一Kafka中（一般JSON）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

原文地址:
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.chinasem.cn/article/1135327。

相关文章

Spring @Scheduled注解及工作原理

Spring @Scheduled注解及工作原理

《Spring@Scheduled注解及工作原理》Spring的@Scheduled注解用于标记定时任务,无需额外库,需配置@EnableScheduling,设置fixedRate、fixedDe... 目录1.@Scheduled注解定义2.配置 @Scheduled2.1 开启定时任务支持2.2 创建
阅读更多...
SpringBoot中使用Flux实现流式返回的方法小结

SpringBoot中使用Flux实现流式返回的方法小结

《SpringBoot中使用Flux实现流式返回的方法小结》文章介绍流式返回（StreamingResponse）在SpringBoot中通过Flux实现,优势包括提升用户体验、降低内存消耗、支持长连... 目录背景流式返回的核心概念与优势1. 提升用户体验2. 降低内存消耗3. 支持长连接与实时通信在Sp
阅读更多...
Spring Boot 实现 IP 限流的原理、实践与利弊解析

Spring Boot 实现 IP 限流的原理、实践与利弊解析

《SpringBoot实现IP限流的原理、实践与利弊解析》在SpringBoot中实现IP限流是一种简单而有效的方式来保障系统的稳定性和可用性,本文给大家介绍SpringBoot实现IP限... 目录一、引言二、IP 限流原理2.1 令牌桶算法2.2 漏桶算法三、使用场景3.1 防止恶意攻击3.2 控制资源
阅读更多...
Mac系统下卸载JAVA和JDK的步骤

Mac系统下卸载JAVA和JDK的步骤

《Mac系统下卸载JAVA和JDK的步骤》JDK是Java语言的软件开发工具包,它提供了开发和运行Java应用程序所需的工具、库和资源,：本文主要介绍Mac系统下卸载JAVA和JDK的相关资料,需... 目录1. 卸载系统自带的 Java 版本检查当前 Java 版本通过命令卸载系统 Java2. 卸载自定
阅读更多...
springboot下载接口限速功能实现

springboot下载接口限速功能实现

《springboot下载接口限速功能实现》通过Redis统计并发数动态调整每个用户带宽,核心逻辑为每秒读取并发送限定数据量,防止单用户占用过多资源,确保整体下载均衡且高效,本文给大家介绍spring... 目录 一、整体目标 二、涉及的主要类/方法✅ 三、核心流程图解（简化） 四、关键代码详解1️⃣ 设置
阅读更多...
Java Spring ApplicationEvent 代码示例解析

Java Spring ApplicationEvent 代码示例解析

《JavaSpringApplicationEvent代码示例解析》本文解析了Spring事件机制,涵盖核心概念（发布-订阅/观察者模式）、代码实现（事件定义、发布、监听）及高级应用（异步处理、... 目录一、Spring 事件机制核心概念1. 事件驱动架构模型2. 核心组件二、代码示例解析1. 事件定义
阅读更多...
SpringMVC高效获取JavaBean对象指南

SpringMVC高效获取JavaBean对象指南

《SpringMVC高效获取JavaBean对象指南》SpringMVC通过数据绑定自动将请求参数映射到JavaBean,支持表单、URL及JSON数据,需用@ModelAttribute、@Requ... 目录Spring MVC 获取 JavaBean 对象指南核心机制：数据绑定实现步骤1. 定义 Ja
阅读更多...
javax.net.ssl.SSLHandshakeException:异常原因及解决方案

javax.net.ssl.SSLHandshakeException:异常原因及解决方案

《javax.net.ssl.SSLHandshakeException:异常原因及解决方案》javax.net.ssl.SSLHandshakeException是一个SSL握手异常,通常在建立SS... 目录报错原因在程序中绕过服务器的安全验证注意点最后多说一句报错原因一般出现这种问题是因为目标服务器
阅读更多...
CSS place-items: center解析与用法详解

CSS place-items: center解析与用法详解

《CSSplace-items:center解析与用法详解》place-items:center;是一个强大的CSS简写属性，用于同时控制网格(Grid)和弹性盒(Flexbox)... place-items: center; 是一个强大的 css 简写属性，用于同时控制 网格(Grid) 和 弹性盒(F
阅读更多...
Java实现删除文件中的指定内容

Java实现删除文件中的指定内容

《Java实现删除文件中的指定内容》在日常开发中,经常需要对文本文件进行批量处理,其中,删除文件中指定内容是最常见的需求之一,下面我们就来看看如何使用java实现删除文件中的指定内容吧... 目录1. 项目背景详细介绍2. 项目需求详细介绍2.1 功能需求2.2 非功能需求3. 相关技术详细介绍3.1 Ja
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2