Flink mongo Kafka

2024-06-07 20:12
文章标签 flink kafka mongo

本文主要是介绍Flink mongo Kafka，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Apache Flink 是一个流处理和批处理的开源平台，用于在分布式环境中处理无界和有界数据流。它提供了用于数据处理的数据流 API（DataStream API）和表 API（Table API），并可以与各种外部数据源和存储系统进行交互。
MongoDB 是一个基于文档的 NoSQL 数据库，它提供了高性能、可扩展和灵活的数据存储。而 Apache Kafka 是一个流处理平台，它允许发布和订阅记录流，类似于消息队列或企业消息系统。
当 Flink 与 MongoDB 和 Kafka 结合使用时，可以构建强大的数据处理管道，用于实时数据流分析和批处理任务。以下是这些组件结合使用时可能的一些用途：

  1. Flink 与 Kafka：
  • Flink 可以作为 Kafka 的消费者（Consumer），从 Kafka 主题（Topics）中读取数据流，并对其进行实时处理。
  • Flink 也可以将数据写入 Kafka，使其成为一个中间存储或数据传递的桥梁。
  • 通过 Flink 的时间窗口和状态管理等特性，可以对 Kafka 中的数据流进行复杂的实时分析。
  1. Flink 与 MongoDB：
  • Flink 可以从 MongoDB 中读取数据，用于批处理或实时分析。
  • Flink 也可以将处理后的数据写入 MongoDB，用于持久化存储或进一步的数据分析。
    使用 Flink 的表 API（Table API）和 SQL 支持，可以方便地对 MongoDB 中的数据进行查询和分析。
  1. Kafka、Flink 和 MongoDB 结合使用：
  • Kafka 可以作为数据源，提供实时数据流给 Flink 进行处理。
  • Flink 对 Kafka 中的数据流进行实时分析，并可能将结果写入 MongoDB 进行存储。
  • MongoDB 中的数据也可以作为 Flink 批处理任务的输入，用于历史数据分析或与其他数据源进行联合分析。

MONGO 2 KAFKA

下面例子是从mongo获取数据插入到kafka：
代码：

public class MongoDBToKafka {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 配置MongoDB源MongoSource<String> mongoSource = MongoSource.<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("wellCastingInfo")
//                .setProjectedFields("_id", "f0", "f1").setFetchSize(2048).setLimit(10000).setNoCursorTimeout(true).setPartitionStrategy(PartitionStrategy.SAMPLE).setPartitionSize(MemorySize.ofMebiBytes(64)).setSamplesPerPartition(10).setDeserializationSchema(new MongoDeserializationSchema<String>() {@Overridepublic String deserialize(BsonDocument document) {return document.toJson();}@Overridepublic TypeInformation<String> getProducedType() {return BasicTypeInfo.STRING_TYPE_INFO;}}).build();// 创建MongoDB数据流DataStream<String> sourceStream = env.fromSource(mongoSource, WatermarkStrategy.noWatermarks(), "kafka Mongo Source");
//        env.fromSource(mongoSource, WatermarkStrategy.noWatermarks(), "MongoDB-Source")
//                .setParallelism(2)
//                .print()
//                .setParallelism(1);// 配置KafkasinkKafkaSink<String> kafkaSink = KafkaSink.<String>builder().setBootstrapServers("localhost:9092")
//                .setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
//                .setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")// 如果你使用String类型的键.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")// 如果你使用byte[]类型的值.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer").setRecordSerializer(KafkaRecordSerializationSchema.builder().setTopic(Constants.TOPIC_NAME).setValueSerializationSchema(new SimpleStringSchema()).build()).setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE).build();// 将数据流写入KafkasourceStream.sinkTo(kafkaSink);// 执行任务env.execute("MongoDB to Kafka");}
}

pom.xml

		<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-mongodb</artifactId></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-base</artifactId></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId></dependency>

KAFKA 2 FILE

从kafka获取数据写入到本地文件
代码：

public class KafkaToWriteText {public static void main(String[] args) throws Exception {// 1. 设置 Flink 执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();String brokers = "localhost:9092";KafkaSource<String> source = KafkaSource.<String>builder().setBootstrapServers(brokers).setTopics(TOPIC_NAME).setGroupId("my-group").setStartingOffsets(OffsetsInitializer.earliest()).setValueOnlyDeserializer(new SimpleStringSchema()).build();DataStreamSource<String> rs = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");// 创建RollingFileSinkString outputPath = "sink.csv";FileSink<String> sink = FileSink.forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8")).withRollingPolicy(DefaultRollingPolicy.builder().withRolloverInterval(Duration.ofMinutes(15)).withInactivityInterval(Duration.ofMinutes(5)).withMaxPartSize(MemorySize.ofMebiBytes(1024)).build()).build();rs.sinkTo(sink);// 6. 执行 Flink 作业env.execute("Kafka Flink Job");}
}

pom.xml

		<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-files</artifactId></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-base</artifactId></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId></dependency>

KAFKA 部署

  1. 下载地址：
    https://downloads.apache.org/kafka/3.7.0/kafka_2.12-3.7.0.tgz
  2. 运行zookeeper
# Start the ZooKeeper service
$ bin/zookeeper-server-start.sh config/zookeeper.properties
  1. 运行kafka
# Start the Kafka broker service
$ bin/kafka-server-start.sh config/server.properties
  1. 验证
# 接受信息
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic TOPIC_WellCastingInfo --from-beginning
# 发送信息
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic TOPIC_WellCastingInfo

这篇关于Flink mongo Kafka的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

如何在一台服务器上使用docker运行kafka集群

如何在一台服务器上使用docker运行kafka集群

《如何在一台服务器上使用docker运行kafka集群》文章详细介绍了如何在一台服务器上使用Docker运行Kafka集群,包括拉取镜像、创建网络、启动Kafka容器、检查运行状态、编写启动和关闭脚本... 目录1.拉取镜像2.创建集群之间通信的网络3.将zookeeper加入到网络中4.启动kafka集群
阅读更多...
IDEA中的Kafka管理神器详解

IDEA中的Kafka管理神器详解

《IDEA中的Kafka管理神器详解》这款基于IDEA插件实现的Kafka管理工具,能够在本地IDE环境中直接运行,简化了设置流程,为开发者提供了更加紧密集成、高效且直观的Kafka操作体验... 目录免安装：IDEA中的Kafka管理神器！简介安装必要的插件创建 Kafka 连接第一步：创建连接第二步：选
阅读更多...
搭建Kafka+zookeeper集群调度

搭建Kafka+zookeeper集群调度

前言 硬件环境 172.18.0.5        kafkazk1        Kafka+zookeeper                Kafka Broker集群 172.18.0.6        kafkazk2        Kafka+zookeeper                Kafka Broker集群 172.18.0.7        kafkazk3
阅读更多...
Flink任务重启策略

Flink任务重启策略

概述 Flink支持不同的重启策略，以在故障发生时控制作业如何重启集群在启动时会伴随一个默认的重启策略，在没有定义具体重启策略时会使用该默认策略。如果在工作提交时指定了一个重启策略，该策略会覆盖集群的默认策略默认的重启策略可以通过 Flink 的配置文件 flink-conf.yaml 指定。配置参数 restart-strategy 定义了哪个策略被使用。常用的重启策略： 固定间隔 (Fixe
阅读更多...
Java消息队列：RabbitMQ与Kafka的集成与应用

Java消息队列：RabbitMQ与Kafka的集成与应用

Java消息队列：RabbitMQ与Kafka的集成与应用 大家好，我是微赚淘客返利系统3.0的小编，是个冬天不穿秋裤，天冷也要风度的程序猿！ 在现代的分布式系统中，消息队列是实现系统间通信、解耦和提高可扩展性的重要组件。RabbitMQ和Kafka是两个广泛使用的消息队列系统，它们各有特点和优势。本文将介绍如何在Java应用中集成RabbitMQ和Kafka，并展示它们的应用场景。 消息队
阅读更多...
Kafka （快速）安装部署

Kafka （快速）安装部署

文章目录 1、软件下载&配置环境1_JDK安装2_Zookeeper安装3_Kafka安装 2、单机安装1_配置主机名和IP映射2_单机Kafka配置 3、集群安装1_配置主机名和IP的映射关系2_时钟同步3_Zookeeper配置信息4_集群Kafka配置 4、kafka的其他脚本命令 1、软件下载&配置环境 下面的操作无论是单机部署还是分布式集群环境下都是通用的。 准
阅读更多...
Kafka 分布式消息系统详细介绍

Kafka 分布式消息系统详细介绍

Kafka 分布式消息系统 一、Kafka 概述1.1 Kafka 定义1.2 Kafka 设计目标1.3 Kafka 特点 二、Kafka 架构设计2.1 基本架构2.2 Topic 和 Partition2.3 消费者和消费者组2.4 Replica 副本 三、Kafka 分布式集群搭建3.1 下载解压3.1.1 上传解压 3.2 修改 Kafka 配置文件3.2.1 修改zookeep
阅读更多...
大数据之Flink（二）

大数据之Flink（二）

4、部署模式 flink部署模式： 会话模式（Session Mode）单作业模式（Per-Job Mode）应用模式（Application Mode） 区别在于集群的生命周期以及资源的分配方式；以及应用的main方法到底在哪里执行。 4.1、会话模式 先启动一个集群，保持一个会话，在这个会话中通过客户端提交作业。集群启动时资源已经确定，所有提交的作业会晶振集群中的资源。适合规模小、执
阅读更多...
Kafka 实战演练：创建、配置与测试 Kafka全面教程

Kafka 实战演练：创建、配置与测试 Kafka全面教程

文章目录 1.配置文件2.消费者1.注解方式2.KafkaConsumer 3.依赖1.注解依赖2.KafkaConsumer依赖 本文档只是为了留档方便以后工作运维，或者给同事分享文档内容比较简陋命令也不是特别全，不适合小白观看，如有不懂可以私信，上班期间都是在得 1.配置文件 Yml配置 spring:kafka:bootstrap-servers: cons
阅读更多...
是谁还不会flink的checkpoint呀~

是谁还不会flink的checkpoint呀~

1、State Vs Checkpoint State:状态,是Flink中某一个Operator在某一个时刻的状态,如maxBy/sum,注意State存的是历史数据/状态,存在内存中。 Checkpoint:快照点, 是Flink中所有有状态的Operator在某一个时刻的State快照信息/存档信息 一句话概括: Checkpoint就是State的快照 目的：假设作业停止了，下次启动的
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2