Flink Kafka获取数据写入到MongoDB中 样例
简述
Apache Flink 是一个流处理和批处理的开源框架,它允许从各种数据源(如 Kafka)读取数据,处理数据,然后将数据写入到不同的目标系统(如 MongoDB)。以下是一个简化的流程,描述如何使用 Flink 从 Kafka 读取数据并保存到 MongoDB:
1、环境准备
- 安装并配置 Apache Flink。
- 安装并配置 Apache Kafka。
- 安装并配置 MongoDB。
- 创建一个 Kafka 主题,并发送一些测试数据。
- 确保 Flink 可以连接到 Kafka 和 MongoDB。
部署参考:
1、flink:Flink 部署执行模式
2、kafka:Flink mongo & Kafka
3、mongoDb:mongo副本集本地部署
2. 添加依赖
在Flink 项目中,需要添加 Kafka 和 MongoDB 的连接器依赖。对于 Maven 项目,可以在 pom.xml 文件中添加相应的依赖。
对于 Kafka,需要添加 Flink Kafka Connector 的依赖。
对于 MongoDB,需要添加 Flink MongoDB Sink 的依赖。
3. 编写 Flink 作业
* 创建一个 Flink 作业,使用 Flink 的 `FlinkKafkaConsumer` 从 Kafka 主题中读取数据。
* 对读取的数据进行必要的转换或处理。
* 使用 MongoDB 的 Java 驱动程序或第三方库将处理后的数据写入 MongoDB。
4. 运行 Flink 作业
使用 Flink 的命令行工具或 IDE 运行 Flink 作业。确保 Kafka 和 MongoDB 正在运行,并且 Flink 可以访问它们。
参考:Flink 命令行提交、展示和取消作业
5. 监控和调试
使用 Flink 的 Web UI 或其他监控工具来监控作业。如果出现问题,检查日志并进行调试。
6. 优化和扩展
根据需求和数据量,优化 Flink 作业的性能和可扩展性。这可能包括调整并行度、增加资源、优化数据处理逻辑等。
代码
package com.wfg.flink.connector.kafka;import com.mongodb.client.model.InsertOneModel;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.mongodb.sink.MongoSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.bson.BsonDocument;import static com.wfg.flink.connector.constants.Constants.KAFKA_BROKERS;
import static com.wfg.flink.connector.constants.Constants.TEST_TOPIC_PV;/*** @author wfg*/
public class KafkaToWriteMongo {public static void main(String[] args) throws Exception {// 1. 设置 Flink 执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();KafkaSource<String> source = KafkaSource.<String>builder().setBootstrapServers(KAFKA_BROKERS).setTopics(TEST_TOPIC_PV).setGroupId("my-test-topic-pv").setStartingOffsets(OffsetsInitializer.latest()).setValueOnlyDeserializer(new SimpleStringSchema()).build();DataStreamSource<String> rs = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");// 创建RollingFileSinkMongoSink<String> sink = MongoSink.<String>builder().setUri("mongodb://root:123456@127.0.0.1:27017,127.0.0.1:27018,127.0.0.1:27019/admin?replicaSet=rs0&authSource=admin").setDatabase("sjzz").setCollection("TestMongoPv").setMaxRetries(3)
// .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE).setSerializationSchema((input, context) -> {System.out.println(input);return new InsertOneModel<>(BsonDocument.parse(input));}).build();rs.sinkTo(sink);// 6. 执行 Flink 作业env.execute("Kafka Flink Job");}
}相关文章:
Flink Kafka获取数据写入到MongoDB中 样例
简述 Apache Flink 是一个流处理和批处理的开源框架,它允许从各种数据源(如 Kafka)读取数据,处理数据,然后将数据写入到不同的目标系统(如 MongoDB)。以下是一个简化的流程,描述如何…...
Android Jetpack Compose入门教程(二)
一、列表和动画 列表和动画在应用内随处可见。在本课中,您将学习如何利用 Compose 轻松创建列表并添加有趣的动画效果。 1、创建消息列表 只包含一条消息的聊天略显孤单,因此我们将更改对话,使其包含多条消息。您需要创建一个可显示多条消…...
如何避免接口重复请求(axios推荐使用AbortController)
前言: 我们日常开发中,经常会遇到点击一个按钮或者进行搜索时,请求接口的需求。 如果我们不做优化,连续点击按钮或者进行搜索,接口会重复请求。 以axios为例,我们一般以以下几种方法为主: 1…...
算法设计与分析:网络流求解棒球赛淘汰问题C++
目录 一、实验目的 二、问题描述 三、实验要求 四、算法思想 1、明显的:win[i]+remain[i][j]<> 2、不明显的:最大流 3、操作 3.1 先读入相关信息(邻接矩阵**k),进行一遍“明显的”判断。 3.2 对剩下的“不明显的”的每个球队构建流网络(邻接表vector< ve…...
Linux Ubuntu 24.04 C语言gcc编译过程详解
下面是Hello World程序源代码文件hello.c的内容,我们将以它为例来说明源文件到可执行文件的形成过程,主要分4步:预处理、汇编、机器码、链接。 #include <stdio.h> int main () {printf ( "hello, world \n " );return 0; }…...
Python自动化办公篇—pandas操作Excel:读取+查看+选择+清洗+排序+筛选+函数+写入
目录 专栏导读库的介绍库的安装1、读取数据2、查看数据3、选择数据4、数据清洗5、数据排序6、数据筛选7、数据操作8、数据写入总结 专栏导读 文章名称链接Python自动化办公—pyautogui图像定位\点击功能,实现自动截取当前屏幕并检索点击(可制作为游戏点击脚本)点我进行跳转Pyt…...
数据库大作业——音乐平台数据库管理系统
W...Y的主页😊 代码仓库分享💕 《数据库系统》课程设计 :流行音乐管理平台数据库系统(本数据库大作业使用软件sql server、dreamweaver、power designer) 目录 系统需求设计 数据库概念结构设计 实体分析 属性分…...
【DBA早下班系列】—— 并行SQL/慢SQL 问题该如何高效收集诊断信息
1. 前言 OceanBase论坛问答区或者提交工单支持的时候大部分时间都浪费在了诊断信息的获取交互上,今天我就其中大家比较头疼的SQL问题,给大家讲解一下如何一键收集并行SQL/慢SQL所需要的诊断信息,减少沟通成本,让大家早下班。 2. …...
用python实现多文件多文本替换功能
用python实现多文件多文本替换功能 今天修改单位项目代码时由于改变了一个数据结构名称,结果有几十个文件都要修改,一个个改实在太麻烦,又没有搜到比较靠谱的工具软件,于是干脆用python手撸了一个小工具,发现python在…...
【DevOps】深入探索Ubuntu操作系统:全面了解
引言 在开源软件的世界里,Ubuntu是一个闪耀的明星。它不仅是一个操作系统,更是一种社区精神、一种共享和协作的文化。Ubuntu操作系统基于强大的Linux内核,由世界各地的开发者共同维护和改进。在这篇博文中,我们将深入探索Ubuntu操…...
【Linux】—MySQL安装
文章目录 前言一、下载官方MySQL包二、下载完成后,通过xftp6上传到Linux服务器上三、解压MySQL安装包四、在安装目录下执行rpm安装,请按顺序依次执行。五、配置MySQL六、启动MySQL数据库七、退出,重新登录数据库 前言 本文主要介绍在Linux环境…...
【vue】form表单提交validate验证不进valid原因
目录 1. 原因 1. 原因 1.<el-form>是否写了ref“form”。2.是否有其它标签写了ref“form”。3.<el-form>中要写成:model,不能使用v-model。4.自定义的validate要各个路径均能返回callback()。 const validatePass (rule, value, callback) > {if (…...
如何用 Google Chrome 浏览器浏览经过 XSLT 渲染的 XML 文件
对于经过XSLT渲染的XML文件,本来,可以直接用 IE (Internet Explorer) 打开,就能看到渲染之后的样子,很方便。但是后来,微软把 IE 换成了 Microsoft Edge,按理说这是比 IE 更先进的浏览器,可是偏…...
Python学习笔记12:进阶篇(二),类的继承与组合
类的继承 我们在编写一系列的类的时候,会发现这些类很相似,但是又有各自的特点和行为。在编写这些类的时候,我们可以把相同的部分抽象成一个基类,然后根据其他不同的特点和行为,抽象出子类,继承这个基类。…...
npm install cnpm -g 报错4048
npm install cnpm -g 报错4048 设置淘宝镜像: 报错如下: 其他博主提供的方法都尝试了,比如管理员权限打开终端,删除.npmrc文件,清除缓存npm cache clean -f等都试了无效,最后怀疑是npm和cnpm版本不对应&…...
本地快速部署 SuperSonic
本地快速部署 SuperSonic 0. 引言1. 本地快速部署 supersonic2. 访问 supersonic3. 支持的数据库4. github 地址 0. 引言 SuperSonic融合Chat BI(powered by LLM)和Headless BI(powered by 语义层)打造新一代的BI平台。这种融合确…...
如何给vue开发的网站做seo?
最近公司有个需求,需要给公司的官网sqlynx做seo,但因为各种历史原因吧,原来的网站是用vue开发的。没办法,只能尝试尽量做一些seo,让网站能更好一些。 目录 1. 服务器端渲染(SSR) 2. 预渲染&am…...
算法训练营第六十天(延长12天添加图论) | LeetCode 647 回文子串、LeetCode 516 最长回文子序列
LeetCode 67 回文子串 思路很简单,每一个dp[i]等于dp[i-1]加上当前字符向前直到0各个长度字符串回文串个数即可 代码如下: class Solution {public boolean isValid(String s) {int l 0, r s.length() - 1;while (l < r) {if (s.charAt(l) ! s.ch…...
TikTok账号养号的流程分享
对于很多刚开始运营TikTok的新手小白来说,都会有一个同样的疑问,那就是:TikTok到底需不需要养号?这里明确告诉大家是需要养号的,今天就把我自己实操过的养号经验和策略总结出来,分享给大家。 一、什么是Ti…...
C++初学者指南第一步---6.枚举和枚举类
C初学者指南第一步—6.枚举和枚举类 文章目录 C初学者指南第一步---6.枚举和枚举类1.作用域的枚举(enum class类型)(C11)2.无作用域的枚举(enum类型)3.枚举类的基础类型4.自定义枚举类映射5.和基础类型的互相转换 1.作用域的枚举(enum class类…...
【JVM】- 内存结构
引言 JVM:Java Virtual Machine 定义:Java虚拟机,Java二进制字节码的运行环境好处: 一次编写,到处运行自动内存管理,垃圾回收的功能数组下标越界检查(会抛异常,不会覆盖到其他代码…...
C++ 求圆面积的程序(Program to find area of a circle)
给定半径r,求圆的面积。圆的面积应精确到小数点后5位。 例子: 输入:r 5 输出:78.53982 解释:由于面积 PI * r * r 3.14159265358979323846 * 5 * 5 78.53982,因为我们只保留小数点后 5 位数字。 输…...
【论文阅读28】-CNN-BiLSTM-Attention-(2024)
本文把滑坡位移序列拆开、筛优质因子,再用 CNN-BiLSTM-Attention 来动态预测每个子序列,最后重构出总位移,预测效果超越传统模型。 文章目录 1 引言2 方法2.1 位移时间序列加性模型2.2 变分模态分解 (VMD) 具体步骤2.3.1 样本熵(S…...
【无标题】路径问题的革命性重构:基于二维拓扑收缩色动力学模型的零点隧穿理论
路径问题的革命性重构:基于二维拓扑收缩色动力学模型的零点隧穿理论 一、传统路径模型的根本缺陷 在经典正方形路径问题中(图1): mermaid graph LR A((A)) --- B((B)) B --- C((C)) C --- D((D)) D --- A A -.- C[无直接路径] B -…...
DingDing机器人群消息推送
文章目录 1 新建机器人2 API文档说明3 代码编写 1 新建机器人 点击群设置 下滑到群管理的机器人,点击进入 添加机器人 选择自定义Webhook服务 点击添加 设置安全设置,详见说明文档 成功后,记录Webhook 2 API文档说明 点击设置说明 查看自…...
人工智能--安全大模型训练计划:基于Fine-tuning + LLM Agent
安全大模型训练计划:基于Fine-tuning LLM Agent 1. 构建高质量安全数据集 目标:为安全大模型创建高质量、去偏、符合伦理的训练数据集,涵盖安全相关任务(如有害内容检测、隐私保护、道德推理等)。 1.1 数据收集 描…...
tomcat指定使用的jdk版本
说明 有时候需要对tomcat配置指定的jdk版本号,此时,我们可以通过以下方式进行配置 设置方式 找到tomcat的bin目录中的setclasspath.bat。如果是linux系统则是setclasspath.sh set JAVA_HOMEC:\Program Files\Java\jdk8 set JRE_HOMEC:\Program Files…...
Python 高效图像帧提取与视频编码:实战指南
Python 高效图像帧提取与视频编码:实战指南 在音视频处理领域,图像帧提取与视频编码是基础但极具挑战性的任务。Python 结合强大的第三方库(如 OpenCV、FFmpeg、PyAV),可以高效处理视频流,实现快速帧提取、压缩编码等关键功能。本文将深入介绍如何优化这些流程,提高处理…...
若依登录用户名和密码加密
/*** 获取公钥:前端用来密码加密* return*/GetMapping("/getPublicKey")public RSAUtil.RSAKeyPair getPublicKey() {return RSAUtil.rsaKeyPair();}新建RSAUti.Java package com.ruoyi.common.utils;import org.apache.commons.codec.binary.Base64; im…...
[USACO23FEB] Bakery S
题目描述 Bessie 开了一家面包店! 在她的面包店里,Bessie 有一个烤箱,可以在 t C t_C tC 的时间内生产一块饼干或在 t M t_M tM 单位时间内生产一块松糕。 ( 1 ≤ t C , t M ≤ 10 9 ) (1 \le t_C,t_M \le 10^9) (1≤tC,tM≤109)。由于空间…...
