Flink最后一站___Flink数据写入Kafka+从Kafka存入Mysql

2023-11-24 09:10
文章标签 数据 mysql flink 写入 kafka database 一站 存入

本文主要是介绍Flink最后一站___Flink数据写入Kafka+从Kafka存入Mysql，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

前言

大家好,我是ChinaManor,直译过来就是中国码农的意思,我希望自己能成为国家复兴道路的铺路人,大数据领域的耕耘者,平凡但不甘于平庸的人。

在这里插入图片描述

今天为大家带来Flink的一个综合应用案例:Flink数据写入Kafka+从Kafka存入Mysql
在这里插入图片描述
第一部分:写数据到kafka中

 public static void writeToKafka() throws Exception{Properties props = new Properties();props.put("bootstrap.servers", BROKER_LIST);props.put("key.serializer", CONST_SERIALIZER);props.put("value.serializer", CONST_SERIALIZER);KafkaProducer<String, String> producer = new KafkaProducer<>(props);//构建User对象，在name为data后边加个随机数int randomInt = RandomUtils.nextInt(1, 100000);User user = new User();user.setName("data" + randomInt);user.setId(randomInt);//转换成JSONString userJson = JSON.toJSONString(user);//包装成kafka发送的记录ProducerRecord<String, String> record = new ProducerRecord<String, String>(TOPIC_USER, partition,null, userJson);//发送到缓存producer.send(record);System.out.println("向kafka发送数据:" + userJson);//立即发送producer.flush();}

重点:

//发送到缓存producer.send(record);

为了增强代码的Robust,我们将常量单独拎出来:

   //本地的kafka机器列表public static final String BROKER_LIST = "192.168.88.161:9092";//kafka的topicpublic static final String TOPIC_USER = "USER";//kafka的partition分区public static final Integer partition = 0;//序列化的方式public static final String CONST_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";//反序列化public static final String CONST_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer";

main方法如下:

public static void main(String[] args) {while(true) {try {//每三秒写一条数据TimeUnit.SECONDS.sleep(3);writeToKafka();} catch (Exception e) {e.printStackTrace();}}}

第二部分:从kafka获取数据

KafkaRickSourceFunction.java

import com.hy.flinktest.entity.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
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.common.TopicPartition;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Properties;@Slf4j
public class KafkaRickSourceFunction extends RichSourceFunction<String>{//kafkaprivate static Properties prop = new Properties();private boolean running = true;//作静态化处理,增强robustprivate static Integer partition = WritedatatoKafka.partition;static {prop.put("bootstrap.servers",WritedatatoKafka.BROKER_LIST);prop.put("zookeeper.connect","192.168.88.161:2181");prop.put("group.id",WritedatatoKafka.TOPIC_USER);prop.put("key.deserializer",WritedatatoKafka.CONST_DESERIALIZER);prop.put("value.deserializer",WritedatatoKafka.CONST_DESERIALIZER);prop.put("auto.offset.reset","latest");prop.put("max.poll.records", "500");prop.put("auto.commit.interval.ms", "1000");}@Overridepublic void run(SourceContext sourceContext) throws Exception {//创建一个消费者客户端实例KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<String, String>(prop);//只消费TOPIC_USER 分区TopicPartition topicPartition = new TopicPartition(WritedatatoKafka.TOPIC_USER,partition);long offset =0; //这个初始值应该从zk或其他地方获取offset = placeOffsetToBestPosition(kafkaConsumer, offset, topicPartition);while (running){ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);if(records.isEmpty()){continue;}for (ConsumerRecord<String, String> record : records) {//record.offset();//record.key()String value = record.value();sourceContext.collect(value);}}}

然后 返回最合适的offset
在这里插入图片描述

    /*** 将offset定位到最合适的位置，并返回最合适的offset。* @param kafkaConsumer consumer* @param offset offset* @param topicPartition partition* @return the best offset*/private long placeOffsetToBestPosition(KafkaConsumer<String, String> kafkaConsumer,long offset, TopicPartition topicPartition) {List<TopicPartition> partitions = Collections.singletonList(topicPartition);kafkaConsumer.assign(partitions);long bestOffset = offset;if (offset == 0) {log.info("由于offset为0，重新定位offset到kafka起始位置.");kafkaConsumer.seekToBeginning(partitions);} else if (offset > 0) {kafkaConsumer.seekToBeginning(partitions);long startPosition = kafkaConsumer.position(topicPartition);kafkaConsumer.seekToEnd(partitions);long endPosition = kafkaConsumer.position(topicPartition);if (offset < startPosition) {log.info("由于当前offset({})比kafka的最小offset({})还要小，则定位到kafka的最小offset({})处。",offset, startPosition, startPosition);kafkaConsumer.seekToBeginning(partitions);bestOffset = startPosition;} else if (offset > endPosition) {log.info("由于当前offset({})比kafka的最大offset({})还要大，则定位到kafka的最大offset({})处。",offset, endPosition, endPosition);kafkaConsumer.seekToEnd(partitions);bestOffset = endPosition;} else {kafkaConsumer.seek(topicPartition, offset);}}return bestOffset;}@Overridepublic void cancel() {running = false;}}

第三部分
主类:从kafka读取数据写入mysql

    //1.构建流执行环境 并添加数据源

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> dataStreamSource = env.addSource(new KafkaRickSourceFunction());

    //2.从kafka里读取数据，转换成User对象

 DataStream<User> dataStream = dataStreamSource.map(lines -> JSONObject.parseObject(lines, User.class));

//3.收集5秒钟的总数

dataStream.timeWindowAll(Time.seconds(5L)).apply(new AllWindowFunction<User, List<User>, TimeWindow>() {@Overridepublic void apply(TimeWindow timeWindow, Iterable<User> iterable, Collector<List<User>> out) throws Exception {List<User> users = Lists.newArrayList(iterable);if(users.size() > 0) {System.out.println("5秒内总共收到的条数：" + users.size());out.collect(users);}}})//sink 到数据库.addSink(new MysqlRichSinkFunction());//打印到控制台//.print();

第四部分:
写入到目标数据库sink
MysqlRichSinkFunction.java

@Slf4j
public class MysqlRichSinkFunction extends RichSinkFunction<List<User>> {private Connection connection = null;private PreparedStatement ps = null;@Overridepublic void open(Configuration parameters) throws Exception {// super.open(parameters);log.info("获取数据库连接");connection = DbUtil.getConnection();String sql = "insert into user1(id,name) values (?,?)";ps = connection.prepareStatement(sql);}public void invoke(List<User> users, Context ctx) throws Exception {//获取ReadMysqlResoure发送过来的结果for(User user : users) {ps.setLong(1, user.getId());ps.setString(2, user.getName());ps.addBatch();}//一次性写入int[] count = ps.executeBatch();log.info("成功写入Mysql数量：" + count.length);}@Overridepublic void close() throws Exception {//关闭并释放资源if(connection != null) {connection.close();}if(ps != null) {ps.close();}}}

总结

以上便是Flink数据写入Kafka+从Kafka存入Mysql
如果有帮助,给manor一键三连吧~~

在这里插入图片描述

这篇关于Flink最后一站___Flink数据写入Kafka+从Kafka存入Mysql的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/421645。

相关文章

大模型研发全揭秘：客服工单数据标注的完整攻略

大模型研发全揭秘：客服工单数据标注的完整攻略

在人工智能（AI）领域，数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者，掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中，工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注，不仅能够帮助提升客服自动化系统的智能化水平，还能优化客户服务流程，提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行
阅读更多...
SQL中的外键约束

SQL中的外键约束

外键约束用于表示两张表中的指标连接关系。外键约束的作用主要有以下三点： 1.确保子表中的某个字段（外键）只能引用父表中的有效记录2.主表中的列被删除时，子表中的关联列也会被删除3.主表中的列更新时，子表中的关联元素也会被更新 子表中的元素指向主表 以下是一个外键约束的实例展示
阅读更多...
基于MySQL Binlog的Elasticsearch数据同步实践

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展，我们的业务数据越来越多，单纯使用 MySQL 已经不能满足我们的数据查询需求，例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后，随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步，我们目前的解决方案是建立数据中间表。把需要检索的业务数据，统一放到一张M
阅读更多...
关于数据埋点，你需要了解这些基本知识

关于数据埋点，你需要了解这些基本知识

产品汪每天都在和数据打交道，你知道数据来自哪里吗？ 移动app端内的用户行为数据大多来自埋点，了解一些埋点知识，能和数据分析师、技术侃大山，参与到前期的数据采集，更重要是让最终的埋点数据能为我所用，否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式，可以区分为： 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理：自动程度高的，能解决通用统计，便于统一化管理，但个性化定
阅读更多...
如何去写一手好SQL

如何去写一手好SQL

MySQL性能 最大数据量 抛开数据量和并发数，谈性能都是耍流氓。MySQL没有限制单表最大记录数，它取决于操作系统对文件大小的限制。 《阿里巴巴Java开发手册》提出单表行数超过500万行或者单表容量超过2GB，才推荐分库分表。性能由综合因素决定，抛开业务复杂度，影响程度依次是硬件配置、MySQL配置、数据表设计、索引优化。500万这个值仅供参考，并非铁律。 博主曾经操作过超过4亿行数据
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
异构存储（冷热数据分离）

异构存储（冷热数据分离）

异构存储主要解决不同的数据，存储在不同类型的硬盘中，达到最佳性能的问题。 异构存储Shell操作 （1）查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies （2）为指定路径（数据存储目录）设置指定的存储策略 hdfs storagepolicies -setStoragePo
阅读更多...
Hadoop集群数据均衡之磁盘间数据均衡

Hadoop集群数据均衡之磁盘间数据均衡

生产环境，由于硬盘空间不足，往往需要增加一块硬盘。刚加载的硬盘没有数据时，可以执行磁盘数据均衡命令。（Hadoop3.x新特性） plan后面带的节点的名字必须是已经存在的，并且是需要均衡的节点。 如果节点不存在，会报如下错误： 如果节点只有一个硬盘的话，不会创建均衡计划： （1）生成均衡计划 hdfs diskbalancer -plan hadoop102 （2）执行均衡计划 hd
阅读更多...
性能分析之MySQL索引实战案例

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题，本文将一起从代码层去分析为什么没有建立索引？ 开源ERP项目地址：https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置
阅读更多...
MySQL数据库宕机，启动不起来，教你一招搞定！

MySQL数据库宕机，启动不起来，教你一招搞定！

作者介绍：老苏，10余年DBA工作运维经验，擅长Oracle、MySQL、PG、Mongodb数据库运维（如安装迁移，性能优化、故障应急处理等）公众号：老苏畅谈运维欢迎关注本人公众号，更多精彩与您分享。 MySQL数据库宕机，数据页损坏问题，启动不起来，该如何排查和解决，本文将为你说明具体的排查过程。 查看MySQL error日志 查看 MySQL error日志，排查哪个表（表空间
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2