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类…...
Appium+python自动化(十六)- ADB命令
简介 Android 调试桥(adb)是多种用途的工具,该工具可以帮助你你管理设备或模拟器 的状态。 adb ( Android Debug Bridge)是一个通用命令行工具,其允许您与模拟器实例或连接的 Android 设备进行通信。它可为各种设备操作提供便利,如安装和调试…...
工业安全零事故的智能守护者:一体化AI智能安防平台
前言: 通过AI视觉技术,为船厂提供全面的安全监控解决方案,涵盖交通违规检测、起重机轨道安全、非法入侵检测、盗窃防范、安全规范执行监控等多个方面,能够实现对应负责人反馈机制,并最终实现数据的统计报表。提升船厂…...
【位运算】消失的两个数字(hard)
消失的两个数字(hard) 题⽬描述:解法(位运算):Java 算法代码:更简便代码 题⽬链接:⾯试题 17.19. 消失的两个数字 题⽬描述: 给定⼀个数组,包含从 1 到 N 所有…...
基于服务器使用 apt 安装、配置 Nginx
🧾 一、查看可安装的 Nginx 版本 首先,你可以运行以下命令查看可用版本: apt-cache madison nginx-core输出示例: nginx-core | 1.18.0-6ubuntu14.6 | http://archive.ubuntu.com/ubuntu focal-updates/main amd64 Packages ng…...
pam_env.so模块配置解析
在PAM(Pluggable Authentication Modules)配置中, /etc/pam.d/su 文件相关配置含义如下: 配置解析 auth required pam_env.so1. 字段分解 字段值说明模块类型auth认证类模块,负责验证用户身份&am…...
sqlserver 根据指定字符 解析拼接字符串
DECLARE LotNo NVARCHAR(50)A,B,C DECLARE xml XML ( SELECT <x> REPLACE(LotNo, ,, </x><x>) </x> ) DECLARE ErrorCode NVARCHAR(50) -- 提取 XML 中的值 SELECT value x.value(., VARCHAR(MAX))…...
BCS 2025|百度副总裁陈洋:智能体在安全领域的应用实践
6月5日,2025全球数字经济大会数字安全主论坛暨北京网络安全大会在国家会议中心隆重开幕。百度副总裁陈洋受邀出席,并作《智能体在安全领域的应用实践》主题演讲,分享了在智能体在安全领域的突破性实践。他指出,百度通过将安全能力…...
MySQL 8.0 OCP 英文题库解析(十三)
Oracle 为庆祝 MySQL 30 周年,截止到 2025.07.31 之前。所有人均可以免费考取原价245美元的MySQL OCP 认证。 从今天开始,将英文题库免费公布出来,并进行解析,帮助大家在一个月之内轻松通过OCP认证。 本期公布试题111~120 试题1…...
华为云Flexus+DeepSeek征文|DeepSeek-V3/R1 商用服务开通全流程与本地部署搭建
华为云FlexusDeepSeek征文|DeepSeek-V3/R1 商用服务开通全流程与本地部署搭建 前言 如今大模型其性能出色,华为云 ModelArts Studio_MaaS大模型即服务平台华为云内置了大模型,能助力我们轻松驾驭 DeepSeek-V3/R1,本文中将分享如何…...
代码规范和架构【立芯理论一】(2025.06.08)
1、代码规范的目标 代码简洁精炼、美观,可持续性好高效率高复用,可移植性好高内聚,低耦合没有冗余规范性,代码有规可循,可以看出自己当时的思考过程特殊排版,特殊语法,特殊指令,必须…...
