[AIGC] flink sql 消费kafka消息,然后写到mysql中的demo

这是一个使用 Flink SQL 从 Kafka 中消费数据并写入 MySQL 的示例。在这个示例中,我们将假设有一个 Kafka 主题 “input_topic”,它产生格式为 (user_id: int, item_id: int, behavior: string, timestamp: long) 的数据,我们需要把这些数据写入名为"output_table"的 MySQL 表中。

首先,我们需要创建表达 Kafka 和 MySQL 的源和目标表条目:

CREATE TABLE kafka_source (user_id INT,item_id INT,behavior STRING,timestamp AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` / 1000, 'yyyy-MM-dd HH:mm:ss')),WATERMARK FOR timestamp as timestamp - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka','topic' = 'input_topic','scan.startup.mode' = 'earliest-offset','properties.bootstrap.servers' = 'localhost:9092','format' = 'json'
);CREATE TABLE mysql_sink (user_id INT,item_id INT,behavior STRING,timestamp TIMESTAMP(3)
) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://localhost:3306/test','table-name' = 'output_table','username' = 'test','password' = '123456'
);

进行消费 Kafka 和写入 MySQL 的操作:

INSERT INTO mysql_sink SELECT * FROM kafka_source;

首先,运行上述 SQL 创建对应 Kafka 和 MySQL 的表。其次,运行 INSERT INTO 语句进行从 Kafka 消费数据并将结果写入 MySQL 表的操作。这是一个基本的操作,你可以根据自己的需求进行相应的修改。

注意,需要根据实际的 Kafka 和 MySQL 的配置,如地址、用户名和密码,来修改上述 SQL 语句中的对应部分。

另外,Flink SQL 对 SQL 语句的语法进行了些许改变以适应流处理的特性,如在 kafka_source 表中的 WATERMARK 和 timestamp 的定义。

这个例子中,你需要确保已经引入了 flink-connector-kafka_2.11、flink-connector-jdbc_2.11 和 mysql-connector-java 等相关的 jar 包依赖。

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

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

ROS2入门21讲__第07讲__节点:机器人的工作细胞

目录 前言 通信模型 案例一:Hello World节点(面向过程) 运行效果 代码解析 创建节点流程 案例二:Hello World节点(面向对象) 运行效果 代码解析 创建节点流程 案例三:物体识别节点 …

民国漫画杂志《时代漫画》第24期.PDF

时代漫画24.PDF: https://url03.ctfile.com/f/1779803-1248635000-177187?p9586 (访问密码: 9586) 《时代漫画》的杂志在1934年诞生了,截止1937年6月战争来临被迫停刊共发行了39期。 ps: 资源来源网络!

GMSL2硬件设计V1.1

一、说明 GMSL(Gigabit Multimedia Serial Links),中文名称为千兆多媒体串行链路,是Maxim公司(现属于ADI)推出的一种高速串行接口,通过同轴电缆或屏蔽双绞线(STP)传输高速串行数据,用于汽车摄像头和显示器应用。GMSL2就是指ADI专有的第二代千兆多媒体串行链路技术,传输…

C++的哈希 哈希表 哈希桶

目录 Unordered系列关联式容器 什么是哈希 哈希表 闭散列 载荷因子α 扩容 查找 删除 字符串哈希算法 最终代码 开散列 插入 查找 删除 最终代码 完整代码 Unordered系列关联式容器 C98中,STL提供了底层为红黑树结构的一系列关联式容器&#xff0…

【数据结构】排序算法大全(快速、堆、归并、插入、折半、希尔、冒泡、计数、基数)各算法比较、解析+完整代码

文章目录 八、排序1.插入排序1.1 直接插入排序1.2 折半插入排序1.3 希尔排序 2.交换排序2.1 冒泡排序2.2 快速排序 3.选择排序3.1 简单选择排序3.2 堆3.2.1 堆排序3.2.2 堆插入删除*完善代码 堆 4.归并、基数、计数排序4.1 归并排序4.2 基数排序4.3 计数排序 5.内部排序算法的比…

基于遗传优化的Sugeno型模糊控制器设计matlab仿真

目录 1.课题概述 2.系统仿真结果 3.核心程序与模型 4.系统原理简介 5.完整工程文件 1.课题概述 基于遗传优化的Sugeno型模糊控制器设计matlab仿真,通过遗传优化算法优化模糊控制器的隶属函数参数,从而获得较优的控制效果。 2.系统仿真结果 3.核心程序与模型 …

如何将前端项目打包并部署到不同服务器环境

学习源码可以看我的个人前端学习笔记 (github.com):qdxzw/frontlearningNotes 觉得有帮助的同学,可以点心心支持一下哈(笔记是根据b站尚硅谷的前端讲师【张天禹老师】整理的,用于自己复盘,有需要学习的可以去b站学习原版视频&…

新书推荐:7.3 for语句

本节必须掌握的知识点: 示例二十四 代码分析 汇编解析 7.3.1 示例二十四 ■for语句语法形式: for(表达式1;表达式2;表达式3) { 语句块; } ●语法解析: 第一步:执行表达式1,表达式1初始化循环变量; …

基于PHP的物业管理的设计与实现

第1章 绪论... 1 1.1 研究背景与意义... 1 1.2 国内外发展现状... 2 第2章 关键技术介绍... 3 2.1 PHP语言... 3 2.2 MySQL数据库... 3 2.3 Zend框架... 4 2.4 B/S架构... 4 第3章 系统需求分析... 5 3.1 可行性分析... 5 3.1.1 技术可行性分析... 5 3.1.2 经济可行…

IDEA 2024.1安装与破解

一、下载 官网地址:https://www.jetbrains.com/idea/download/other.html 二、安装 傻瓜式安装即可 三、破解 3.1 破解程序 网站:https://3.jetbra.in/ 3.2 获取激活码 点击*号部分即可复制成功

IOC控制反转

IOC IOC,全称为Inversion of Control(控制反转),是一种设计原则,它反转了传统编程中的控制流程。在传统的编程模式中,组件之间的依赖关系是由组件自身在内部创建和维护的。而在控制反转模式中,这种依赖关系由外部容器(…

【Docker实战】进入四大数据库的命令行模式

上一篇我们讲了docker exec命令,这一次我们使用docker exec命令来进入四大数据库的命令行模式。 我们进行游戏开发或软件开发是离不开四大数据库的,这四大数据库分别是关系型数据库mysql、postgres,nosql数据库redis、mongodb。将它们容器化…

云上聚智——移动云云服务器进行后端的搭建及部署

什么是移动云 移动云是指将移动设备和云计算技术相结合,为移动应用提供强大的计算和存储能力的服务模式。传统的移动应用通常在本地设备上进行计算和存储,而移动云将这些任务转移到云端进行处理。通过移动云,移动设备可以利用云端的高性能计算…

【C++】——入门基础知识超详解

​​​​​​​ 目录 ​编辑 1.C关键字 2. 命名空间 2.1 命名空间定义 2.2 命名空间使用 命名空间的使用有三种方式: 注意事项 3. C输入&输出 示例 1:基本输入输出 示例 2:读取多个值 示例 3:处理字符串输入 示例…

神经网络不确定性综述(Part II)——Uncertainty estimation_Single deterministic methods

相关链接: 神经网络不确定性综述(Part I)——A survey of uncertainty in deep neural networks-CSDN博客 神经网络不确定性综述(Part II)——Uncertainty estimation_Single deterministic methods-CSDN博客 神经网络不确定性综述(Part III)——Uncertainty est…

Docker-Android安卓模拟器本地部署并实现远程开发测试

文章目录 1. 虚拟化环境检查2. Android 模拟器部署3. Ubuntu安装Cpolar4. 配置公网地址5. 远程访问小结 6. 固定Cpolar公网地址7. 固定地址访问 本文主要介绍如何在Ubuntu系统使用Docker部署docker-android安卓模拟器,并结合cpolar内网穿透工具实现公网远程访问本地…

用常识滚雪球:拼多多的内生价值,九年的变与不变

2024年5月22日,拼多多公布了今年一季度财报,该季度拼多多集团营收868.1亿元,同比增长131%,利润306.0亿,同比增长了202%,数据亮眼。 市场对拼多多经历了“看不见”、“看不懂”、“跟不上”三个阶段。拼多多…

STM32无源蜂鸣器播放音乐

开发板:野火霸天虎V2 单片机:STM32F407ZGT6 开发软件:MDKSTM32CubeMX 文章目录 前言一、找一篇音乐的简谱二、确定音调三、确定节拍四、使用STM32CubeMX生成初始化代码五、代码分析 前言 本实验使用的是低电平触发的无源蜂鸣器 无源蜂鸣器是…

【数据库】通过一个实例来认识数据流图DFD

导读:通过一个实例(数据中台)说明数据流图DFD的作用、介绍了常见的数据流图元素及其标准符号以及如何画数据流图。数据流图主要被分析师、系统设计师、流程优化专家、系统管理员以及与系统开发和维护相关的人员查看和使用。对于刚考完2024年5…

设计模式 18 迭代器模式 Iterator Pattern

设计模式 18 迭代器模式 Iterator Pattern 1.定义 迭代器模式 (Iterator Pattern) 是一种行为型设计模式,它提供了一种访问集合元素的标准方法,而无需暴露集合的内部表示。 提供一种方法顺序访问一个聚合对象中的各个元素,而又不需要暴露该…