项目整体结构图如下

参考文章或视频链接
[1] Architecture-for-real-time-video-streaming-analytics

一、将摄像头数据发送至kafka

这个代码将运行在你有摄像头的机器上，缺依赖就装依赖

import cv2
import kafka
import numpy as np# 设置 Kafka Producer
# 注意修改你的kafka地址
producer = kafka.KafkaProducer(bootstrap_servers='localhost:9092')# 打开摄像头（0 为默认摄像头）
cap = cv2.VideoCapture(0)while True:# 从摄像头捕获帧ret, frame = cap.read()if not ret:break# 将图像编码为 JPEG 格式_, buffer = cv2.imencode('.jpg', frame)# 将图像作为字节数组发送到 Kafkaproducer.send('camera-images', buffer.tobytes())# 显示当前捕获的帧cv2.imshow('Video', frame)# 按 'q' 键退出if cv2.waitKey(1) & 0xFF == ord('q'):break# 释放资源
cap.release()
cv2.destroyAllWindows()
producer.close()

二、Kafka准备topic

在准备topic之前，要先配置kafka中的config/server.properties文件，否则其它机器无法联通kafka，配置好后重启kafka。

# 找到这两个选项并修改成如下内容
listeners=PLAINTEXT://0.0.0.0:9092
# 改成你的kafka所在服务器ip
advertised.listeners=PLAINTEXT://{your_ip}:9092

如果你之前创建过topic，那就清空这些topic中的数据

# 设置保留时间为0，相当于立即清空数据
#bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name {your_topic_name} --add-config retention.ms=0
# 恢复原始保留设置，立即清空数据后，将数据的保留时间恢复至原有状态
#bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name {your_topic_name} --add-config retention.ms=604800000

开始正式创建topic

# 创建输入图片所在topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic camera-images --partitions 1 --replication-factor 1
# 创建输出的gray灰度图片所在topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic result-gray-images --partitions 1 --replication-factor 1# 准备好后查看下topic list进行验证
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
# 查看某topic中的数据
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic {your_topic_name} --from-beginning

三、spark读取kafka图像数据并处理

首先给你的spark脚本所运行的python环境(这个环境一般可以为conda等虚拟环境)，安装必要的依赖库

pip install opencv-python-headless

准备脚本文件

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import BinaryType
import cv2
import numpy as npbootstrapServers = "localhost:9092"# 创建 SparkSession
spark = SparkSession.builder \.appName("Kafka-Spark-OpenCV") \.getOrCreate()# 初始化 Kafka Producer，用于发送处理后的图像
# 如果不这样做，会出现PicklingError，因为如果UDF中，包含了无法被序列化的对象，例如线程锁（_thread.RLock）或 Kafka 的 KafkaProducer 实例，序列化就会失败。
# 因此，在每个执行器内部，创建 KafkaProducer 实例
producer = None# 从 Kafka 读取数据流
df = spark \.readStream \.format("kafka") \.option("kafka.bootstrap.servers", "localhost:9092") \.option("subscribe", "camera-images") \.load()# UDF 用于将图像转换为灰度
def convert_to_gray(image_bytes):global producer# 创建 KafkaProducer 实例（在每个执行器上只初始化一次）if producer is None:producer = KafkaProducer(bootstrap_servers = bootstrapServers)# 将字节数组转换为 numpy 数组nparr = np.frombuffer(image_bytes, np.uint8)# 将 numpy 数组解码为图像img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)# 将图像转换为灰度gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)# 将灰度图像编码为 JPEG_, buffer = cv2.imencode('.jpg', gray)# 将处理后的图像发送到 Kafka 'result-gray-images' 主题producer.send('result-gray-images', buffer.tobytes())return buffer.tobytes()# 注册 UDF
convert_to_gray_udf = udf(convert_to_gray, BinaryType())# 应用 UDF 对数据进行灰度化处理
gray_df = df.withColumn("gray_image", convert_to_gray_udf("value"))# 将处理后的数据写入文件或其他输出
query = gray_df.writeStream \.outputMode("append") \.format("console") \.start()# query = gray_df\
#     .writeStream \
#     .format('kafka') \
#     .outputMode('update') \
#     .option("kafka.bootstrap.servers", bootstrapServers) \
#     .option('checkpointLocation', '/spark/job-checkpoint') \
#     .option("topic", "result-gray-images") \
#     .start()query.awaitTermination()

用spark-submit提交脚本文件：

# 1.提高内存
# 2.调整 Kafka 批次大小，减少单个批次的数据量，从而降低内存使用(这个步骤存疑)
/opt/spark-3.5.2-bin-hadoop3/bin/spark-submit \
--executor-memory 4g \
--driver-memory 4g \
--conf "spark.kafka.maxOffsetsPerTrigger=1000" \
--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.2,org.apache.kafka:kafka-clients:3.5.2 \
/opt/spark-3.5.2-bin-hadoop3/jobs/pyjobs/kafka_to_spark.py

四、本地显示灰度图像(存在卡顿现象，待优化)

import cv2
import numpy as np
from kafka import KafkaConsumer# 设置 Kafka Consumer
consumer = KafkaConsumer('result-gray-images',bootstrap_servers='{your_kafka_ip}:9092',auto_offset_reset='latest',enable_auto_commit=True,# group_id='image-display-group'
)# 从 Kafka 主题读取灰度图像并显示
for message in consumer:# print("reading gray image.... ")# 将消息转换为 numpy 数组nparr = np.frombuffer(message.value, np.uint8)# 解码为图像gray_img = cv2.imdecode(nparr, cv2.IMREAD_GRAYSCALE)# 显示灰度图像cv2.imshow('Gray Video', gray_img)if cv2.waitKey(1) & 0xFF == ord('q'):break# 释放资源
cv2.destroyAllWindows()
consumer.close()