当前位置：首页 > news >正文

Flink/Kafka在python中的用处

news 来源：原创 2025/7/7 4:08:12

一、基础概念

1. Apache Kafka 是什么？

核心功能：Kafka 是一个分布式流处理平台，主要用于构建实时数据管道和流式应用程序。
核心概念：
- 生产者（Producer）：向 Kafka 发送数据的程序。
- 消费者（Consumer）：从 Kafka 读取数据的程序。
- 主题（Topic）：数据流的分类名称（类似数据库中的表）。
- Broker：Kafka 集群中的单个服务器节点。
用途：
- 实时数据传输（如日志、事件流）。
- 缓冲数据，解耦生产者和消费者。
- 支持高吞吐量、低延迟的消息传递。

2. Apache Flink 是什么？

核心功能：Flink 是一个分布式流处理和批处理框架，擅长处理无界（实时）和有界（离线）数据流。
核心概念：
- DataStream API：用于处理实时数据流。
- 窗口（Window）：将无限数据流切分为有限块进行处理（如统计每分钟的访问量）。
- 状态（State）：在流处理中保存中间计算结果。
用途：
- 实时数据分析（如监控、报警）。
- 复杂事件处理（如检测异常模式）。
- 流式 ETL（数据清洗、转换）。

二、Kafka + Flink 的协同工作

典型架构：

数据源 → Kafka（收集和存储数据流）。
Kafka → Flink（实时消费和处理数据）。
Flink → 数据库/API/存储系统（输出处理结果）。

优势：

解耦：Kafka 作为中间层，缓冲数据并解耦生产者和消费者。
容错：Kafka 持久化数据，Flink 支持故障恢复。
高吞吐：两者均支持分布式处理，适合大数据场景。

三、Python 中的使用场景

虽然 Kafka 和 Flink 的原生 API 主要基于 Java/Scala，但 Python 可以通过以下方式使用它们：

1. Python 与 Kafka

用途：
- 用 Python 编写生产者或消费者，与 Kafka 交互。
- 适用于轻量级数据处理或与其他 Python 生态工具（如 Pandas、TensorFlow）集成。
工具库：
- confluent-kafka：官方推荐的 Python 客户端库。
- kafka-python：另一个常用库（功能稍少，但简单）。

示例：Python 生产者

from confluent_kafka import Producerproducer = Producer({'bootstrap.servers': 'localhost:9092'})def send_message(topic, message):producer.produce(topic, message)producer.flush()send_message('my_topic', 'Hello Kafka from Python!')

示例：Python 消费者

from confluent_kafka import Consumerconsumer = Consumer({'bootstrap.servers': 'localhost:9092','group.id': 'my-group'
})
consumer.subscribe(['my_topic'])while True:msg = consumer.poll(1.0)if msg is not None:print(f'Received: {msg.value()}')

2. Python 与 Flink（PyFlink）

用途：
- 用 Python 编写 Flink 流处理或批处理作业。
- 适合熟悉 Python 的开发者进行快速原型开发。
工具库：
- PyFlink：Flink 的 Python API（需要 Java 环境支持）。

示例：PyFlink 流处理

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment# 创建环境
env = StreamExecutionEnvironment.get_execution_environment()
table_env = StreamTableEnvironment.create(env)# 从 Kafka 读取数据
table_env.execute_sql("""CREATE TABLE kafka_source (message STRING) WITH ('connector' = 'kafka','topic' = 'my_topic','properties.bootstrap.servers' = 'localhost:9092','format' = 'raw')
""")# 处理数据（例如：统计消息长度）
result_table = table_env.sql_query("SELECT message, LENGTH(message) FROM kafka_source")# 输出到控制台
table_env.execute_sql("""CREATE TABLE print_sink (message STRING,length INT) WITH ('connector' = 'print')
""")result_table.execute_insert("print_sink").wait()

四、典型应用场景

1. 实时日志分析

Kafka 收集服务器日志 → Flink 实时统计错误频率 → Python 发送报警邮件。

2. 用户行为分析

Kafka 接收用户点击事件 → Flink 计算实时点击热力图 → Python 可视化展示。

3. 物联网（IoT）数据处理

Kafka 接收传感器数据 → Flink 检测异常温度 → Python 调用控制 API。

五、注意事项

性能限制：Python 在流处理中的性能通常不如 Java/Scala，适合轻量级任务。
环境依赖：PyFlink 需要 Java 环境，且部分高级功能可能受限。
学习曲线：需熟悉 Kafka/Flink 的核心概念（如分区、容错、状态管理）。

六、总结

Kafka：用于可靠地传输和缓冲实时数据。
Flink：用于复杂流处理（窗口、聚合、状态管理）。
Python：通过 confluent-kafka 和 PyFlink 实现轻量级集成。

如果你需要处理大规模实时数据流，且希望用 Python 快速开发，Kafka + Flink 是一个强大的组合！

Flink/Kafka在python中的用处

一、基础概念 1. Apache Kafka 是什么？ 核心功能：Kafka 是一个分布式流处理平台，主要用于构建实时数据管道和流式应用程序。核心概念： 生产者（Producer）：向 Kafka 发送数据的程序。…...

编程日记 2025/7/7 4:08:12

【Spring Boot 与 Spring Cloud 深度 Mape 之十】体系整合、部署运维与进阶展望

【Spring Boot 与 Spring Cloud 深度 Mape 之十】体系整合、部署运维与进阶展望 #微服务实战 #Docker #Kubernetes #SpringSecurity #OAuth2 #分布式事务 #Seata #ServiceMesh #总结 #SpringCloud #SpringBoot 系列终章：经过前九篇 [【深度 Mape 系列】] 的系统学习…...

编程日记 2025/7/7 4:01:42

轻松理解Python装饰器：从基础到应用

一、为什么需要装饰器想象一下，你写了很多函数来完成不同的任务，突然有个新需求：在每个函数执行前打印一条“函数开始执行”的消息，执行后打印“函数执行结束”。如果没有装饰器，你就得在每个函数里手动添加这两条打…...

编程日记 2025/7/7 4:06:53

RabbitMQ 技术详解：异步消息通信的核心原理与实践

这里写目录标题 RabbitMQ 技术详解：异步消息通信的核心原理与实践一、RabbitMQ 本质剖析核心架构组件二、核心功能与应用场景主要作用典型应用场景三、工作流程深度解析消息传递流程关键协议机制四、Java 实现示例1. 依赖配置（Maven）2. 消…...

编程日记 2025/7/4 12:04:55

MySQL-- 多表查询的分类，SQL92与SQL99，7种JOIN的实现，SQL99语法的新特性

目录一，多表查询的分类角度1：等值连接 vs 非等值连接角度2：自连接 vs 非自连接角度3：内连接 vs 外连接二，SQL92语法实现内连接：见上，略SQL92语法实现外连接：使用 -…...

编程日记 2025/7/7 3:37:08

Selenium文件上传

在 Web 自动化测试中，文件上传是一项常见的任务。不同的网站和前端技术可能导致上传方式有所不同，因此需要采用不同的方法进行处理。方法 1：使用 send_keys() 直接上传（最常用）适用场景：页面中有标准的 <input type="file"> 标签。不需要弹出 Wind…...

编程日记 2025/7/7 2:54:49

getID3获取本地或远程视频时长

音频文件也可使用，使用ffmeg安装太复杂了附ffmpeg方式：centos下安装ffmpeg_yum安装ffmpeg-CSDN博客使用composer先安装 composer require james-heinrich/getid3 获取本地视频 //获取本地视频$video_path $_SERVER[DOCUMENT_ROOT].$params[video];…...

编程日记 2025/7/7 3:01:28

OpenAI流式解析

OpenAI 流式的代码： 首选一般请使用os.getenv 去读环境变量的内容注意使用pip install python-dotenv 的安装方法 load_dotenv 是这个库提供的一个函数，用于读取 .env 文件并将其中定义的键值对设置为系统的环境变量。默认情况下，load_…...

编程日记 2025/7/7 3:18:53

在Trae中设置Python解释器版本

Python 是一种广泛使用的高级编程语言，因其简洁易读的语法和强大的功能而备受欢迎。随着 Python 的不断发展，多个版本相继发布，每个版本都带来了新特性和改进。然而，这也带来了一些问题，比如不同的工程，需要…...

编程日记 2025/7/6 21:52:46

第 6 章:优化动态分配内存的变量_《C++性能优化指南》_notes

优化动态分配内存的变量第六章核心知识点详解总结第六章动态内存优化重点难点梳理一、多选题（每题至少2个正确答案）二、设计题答案与详解多选题答案设计题答案示例第六章核心知识点详解动态内存分配的开销知识点：动态内存分配需要调用…...

编程日记 2025/7/3 9:00:38

图像数据增强教程：为目标检测任务准备数据

目录一、简介二、代码结构三、环境要求四、数据增强类 4.1 调整增强概率和参数 4.2 增强方法参数五、数据增强主函数六、主函数效果展示完整代码一、简介在目标检测任务中，数据增强是一种关键技术，通过对原始图像应用多种变换来增加…...

编程日记 2025/7/2 20:07:56

Unity编辑器功能及拓展(2) —Gizmos编辑器绘制功能

Unity中的Gizmos功能是用于在场景视图中绘制辅助图形或图标的工具，帮助开发者在编辑模式下直观调试和可视化游戏对象的位置、范围、方向等信息。一.定义概述 Gizomsd 概述 Gizoms是Unity提供的一个API，或者叫做一个工具类，包含一系列静态…...

编程日记 2025/7/6 15:55:03

HarmonyOS WebSocket全场景应用开发深度解析

注：适用版本（Harmony OS NEXT / 5.0 / API 12 ） 一、最终效果预览二、基础代码结构 Entry Component struct ChatApp {State messages: Message[] [] // 所有聊天记录State inputText: string "" // 输入框内容State isCon…...

编程日记 2025/7/2 22:32:24

JCRQ1河马算法+消融实验！HO-CNN-LSTM-Attention系列四模型多变量时序预测

JCRQ1河马算法消融实验！HO-CNN-LSTM-Attention系列四模型多变量时序预测目录 JCRQ1河马算法消融实验！HO-CNN-LSTM-Attention系列四模型多变量时序预测预测效果基本介绍程序设计参考资料预测效果基本介绍基于HO-CNN-LSTM-Attention、CNN-LSTM-Attent…...

编程日记 2025/7/4 14:35:46

ubuntu 安装 postgresql

在 Ubuntu 系统中安装 PostgreSQL 的步骤如下： 步骤 1：更新软件包列表 sudo apt update步骤 2：安装 PostgreSQL Ubuntu 默认仓库包含 PostgreSQL，直接安装： sudo apt install postgresql postgresql-contrib -ypost…...

编程日记 2025/7/4 1:19:37

深入实践：基于WebSocket的全球化金融数据实时对接方案。马来西亚、印度、美国金融数据API

深入实践：基于WebSocket的全球化金融数据实时对接方案在全球金融市场中，实时数据的高效获取与处理是量化交易、行情监控等场景的核心能力。本文将以技术实践为核心，详细解析如何通过WebSocket技术实现美国、印度、马来西亚等多国金融数据&a…...

编程日记 2025/7/1 9:23:47

深度学习处理时间序列（5）

Keras中的循环层上面的NumPy简单实现对应一个实际的Keras层—SimpleRNN层。不过，二者有一点小区别：SimpleRNN层能够像其他Keras层一样处理序列批量，而不是像NumPy示例中的那样只能处理单个序列。也就是说，它接收形状为(batch_si…...

编程日记 2025/7/5 16:37:13

Linux：进程间通信

目录一前言二进程间通信目的三进程间通信方法四管道通信 1. 进程如何通信 2.管道概念 2.1匿名管道 2.2 匿名管道对多个进程的控制 2.3 命名管道 2.4 命名管道原理一前言在我们学习进程的时候，我们知道正是因为程序地址空间的存在&#xff…...

编程日记 2025/7/3 1:22:21

为什么idea显示数据库连接成功，但操作数据库时，两边数据不同步

今日份小bug又叕又来了！ 一、原因分析 1. 未提交的事务 - IDEA 中执行了修改操作但未提交事务 - 其他客户端有未提交的修改 2. 连接隔离级别问题 - 不同连接使用了不同的事务隔离级别 - 读未提交(READ UNCOMMITTED)导致看到未提交数据 3. 多客户端同时操作…...

编程日记 2025/7/3 21:24:47

VMware中新建Ubuntu虚拟机系统，并安装Anaconda

详细介绍 Ubuntu18.04版本的安装Anaconda的安装 Ubuntu20.04版本的安装给出其他参考安装Ubuntu18.04 新建虚拟机如果不习惯图文形式的，也可参考该up主的环境安装分享，和我如下记录有些不同，但不影响，大部分均一致。 …...

编程日记 2025/7/6 13:35:31

LangChain 基础系列之 Prompt 工程详解：从设计原理到实战模板

LangChain 基础系列之 Prompt 工程详解：从设计原理到实战模板一、揭开 LangChain 的 “灵魂引擎”：Prompt 的核心作用在 LangChain 构建的智能应用中，Prompt（提示词）堪称驱动大模型的 “神经中枢”。这个承载着任务…...

编程日记 2025/7/2 16:11:22

项目如何安装本地tgz包并配置局部registry

一、判断包来源是否正确 1. 检查url curl <registry_url>2. 查看包是否存在 npm view <package_name> --registry<registry_url>二、局部registry配置步骤： 1. 全局配置如果你希望对所有项目生效，可以将这行配置添加到全局.npmr…...

编程日记 2025/7/4 14:34:38

unity客户端面试高频2（自用）

标题是我 1.构造函数为什么不能为虚函数？析构函数为什么要虚函数？2.C智能指针3.左值和右值完美转发 4.深拷贝与浅拷贝5.malloc VS new 你们知道吗 1.构造函数为什么不能为虚函数？析构函数为什么要虚函数？ 构造函数不能定义为虚函…...

编程日记 2025/7/6 17:48:36

【12】Ajax的原理和解析

一、前言二、什么是Ajax 三、Ajax的基本原理 3.1 发送请求 3.2 解析内容 3.3 渲染网页 3.4 总结四、Ajax 分析五、过滤请求-筛选所有Ajax请求一、前言当我们在用 requests 抓取页面的时候，得到的结果可能会和在浏览器中看到的不一样&a…...

编程日记 2025/7/1 23:55:16

深度学习在测距模型中的应用

一、单目视觉测距和双目视觉测距简介 1、单目视觉测距模型：深度估计（Depth Estimation） 原理：通过深度学习模型（如MonoDepth2、MiDaS）或传统的计算机视觉方法（如单目相机结合物体大小推断&am…...

编程日记 2025/7/4 5:04:35

np.vectorize 是 NumPy 提供的一个用于将标量函数（scalar function）向量化的工具，使其可以作用于 NumPy 数组，类似于通用函数（ufunc）的行为。 1️⃣ np.vectorize 语法 numpy.vectorize(pyfunc, otypes=None, signature=None, excluded=None, cache=False)📌 参数…...

编程日记 2025/7/1 4:22:29

HarmonyOS NEXT状态管理实践

在HarmonyOS NEXT开发中，状态管理是构建高效、响应式应用的核心。本文深入探讨状态管理的最佳实践，结合代码示例与案例分析，帮助开发者掌握这一关键技能。一、状态管理装饰器的合理使用 HarmonyOS NEXT提供多种状态管理装饰器，…...

编程日记 2025/6/28 6:26:22

广告牌变“高空炸弹“？智能预警终端筑起安全防线!

近年来，随着城市发展步伐加快，广告牌已成为城市形象的重要载体。但与此同时，因设计缺陷、违规搭建、维护缺失等问题导致的广告牌坠落事故频发，给市民生命财产安全带来严重威胁。据不完全统计，我国2000万块户外广告牌中…...

编程日记 2025/6/28 1:32:15

scss预处理器对比css的优点以及基本的使用

本文主要在vue中演示，scss的基本使用。安装命令 npm install sass sass-loader --save-dev 变量 SCSS 支持变量，可将常用的值（如颜色、字体大小、间距等）定义为变量，方便重复使用和统一修改。 <template><…...

编程日记 2025/6/27 15:00:58

Redis 单线程

Redis 读写是否是单线程？ 核心数据操作仍然是单线程 Redis 主要采用单线程执行命令，这是因为： 避免加锁：如果多个线程并发修改数据，就需要加锁，而 Redis 采用单线程保证操作的原子性，无需加…...

编程日记 2025/7/1 22:38:34

Node.js 下载安装及环境配置教程、卸载删除环境配置超详细步骤(附图文讲解!) 从零基础入门到精通，看完这一篇就够了

Node.js 安装一、进入官网地址下载安装包 Node.js — Download Node.js 选择对应你系统的Node.js版本，这里我选择的是Windows系统、64位 Tips：如果想下载指定版本，点击【以往的版本】，即可选择自己想要的版本下载二、安装程序…...

编程日记 2025/7/5 13:02:00

第十五章:Python的Pandas库详解及常见用法

在数据分析领域，Python的Pandas库是一个不可或缺的工具。它提供了高效的数据结构和数据分析工具，使得数据处理变得简单而直观。本文将详细介绍Pandas库的基本功能、常见用法，并通过示例代码演示如何使用Pandas进行数据处理。最后，…...

编程日记 2025/7/1 20:48:32

Windows下VSCode的安装

前言 VSCode的安装看起来平平无奇，但也不是轻轻松松的。笔者将最新的Windows下安装VSCode，以及运行最简单的C程序的过程记录下来，供后续的自己和大家参考。一、官网下载安装包 Visual Studio Code - Code Editing. Redefined 二、安装直接…...

编程日记 2025/7/3 13:18:30

PgDog：一个PostgreSQL分布式集群中间件

PgDog 是一个实现了 PostgreSQL 分片、连接池以及负载均衡功能的中间。PgDog 使用 Rust 语言编写，支持跨平台（Linux、Mac OS、Windows），具有高性能和高可靠性，可以在不需要修改任何应用程序的前提下实现 PostgreSQL 数…...

编程日记 2025/7/1 4:52:01

基于yolov11的棉花品种分类检测系统python源码+pytorch模型+评估指标曲线+精美GUI界面

【算法介绍】基于YOLOv11的棉花品种分类检测系统是一种高效、准确的农作物品种识别工具。该系统利用YOLOv11深度学习模型，能够实现对棉花主要品种，包括树棉（G. arboreum）、海岛棉（G. barbadense）、草棉&a…...

编程日记 2025/7/5 21:11:56

Web网页内嵌福昕OFD版式办公套件实现在线预览编辑PDF、OFD文档

PDF，即Portable Document Format，用于以一种独立于应用程序、硬件、操作系统的方式共享和查看文档；OFD，即Office Open Document Format for Document，是一种在政府公文和法律文件等领域广泛应用的电子文件格式&#xf…...

编程日记 2025/7/6 10:23:57

UE4学习笔记 FPS游戏制作32 主菜单，暂停游戏，显示鼠标指针

文章目录一主菜单搭建UI显示主菜单时，暂停游戏，显示鼠标绑定按钮二打开主菜单一主菜单搭建UI 添加一个MainUi的控件添加一个返回游戏的按钮和一个退出游戏的按钮修改一下样式，放中间显示主菜单时，暂停游戏&#xff0…...

编程日记 2025/7/3 23:43:44

多线程 - 线程安全引入

写一个代码，让主线程创建一个新的线程，由新的线程负责完成一系列的运算（比如：1 2 3 ... 1000），再由主线程负责获取到最终结果。但打印结果为 result 0，略微思考，明白了要让 t 线…...

编程日记 2025/7/6 11:14:03

Angular项目改端口号

在 Angular 项目中修改开发服务器的端口号（默认是 4200），可以通过以下几种方式实现： 方法 1：通过 ng serve 命令行参数直接在运行 ng serve 时指定端口号： ng serve --port 4300效果：开发服务…...

编程日记 2025/7/3 10:43:25

论文内可解释性分析

目录 3 TEPM(Text-Enhanced Prototype Module)3.1 为什么要进行文本增强？（动机）3.2 为什么要使用 Concat(Fv, T) 和 Repeat(T) + Fv？3.3 为什么 Q=F_C，K=V=F_R ？（第一层注意力）3.4 为什么要进行两层注意力？3.5 为什么最终结果会更好？**3.6 面试官可能问的挑战性问题*…...

编程日记 2025/7/6 22:54:49