pyspark streaming简介和消费 kafka示例

郑文峰的博客

使用dify对接飞书多维表格使用n8n对接飞书多维表格服务启动时出现 OOM 一次服务升级时pg表DDL执行超时失败 Go语言高效IO缓冲技术详解 Go语言延迟初始化(Lazy Initialization)最佳实践 Go语言字符串拼接性能对比与优化指南 Go语言结构体内存对齐完全指南 Go语言空结构体：零内存消耗的高效编程 Go语言堆栈分配与逃逸分析深度解析 Go语言原子操作完全指南 Go语言内存预分配完全指南 Go语言不可变数据共享：无锁并发编程实践 Go语言零拷贝技术完全指南 Go语言遍历性能深度解析：从原理到优化实践 Go语言Interface Boxing原理与性能优化指南 Go协程池深度解析：原理、实现与最佳实践使用etcd分布式锁导致的协程泄露与死锁问题基于pre-commit的Python代码规范落地实践初识 MCP Server pulsar阻塞导致logstash无法接入日志 django-prometheus使用及源码分析 kube-proxy源码分析 kubernetes service如何通过iptables转发 tcp缓存引起的日志丢失 django-apschedule定时任务异常停止理解calico容器网络通信方案原理理解flannel的三种容器网络方案原理理解Linux IPIP隧道理解VXLAN网络理解Linux TunTap设备快速了解iptables kafka中listener和advertised.listeners的作用 django rest_framework 分页 django后端服务、logstash和flink接入VictoriaMetrics指标监控 python中import原理 docker容器单机网络手动实现docker容器bridge网络模型 mysql之MVCC原理 mysql之日志使用java开发logstash的filter插件使用python实现单例模式的三种方式 redis之缓存 redis之分片集群 redis之哨兵机制 redis之主从库同步 redis之持久化 redis之五种基本数据类型 go中如何处理error pod中将代码与运行环境分离 ddt源码分析 python装饰器的使用方法读书笔记:如何阅读一本书使用ddt实现unittest的参数化测试使用kubeadm安装k8s 优化gin表单的错误提示信息 gin中validator模块的源码分析 go简单使用grpc python简单使用grpc k8s之PV、PVC和StorageClass k8s之StatefulSet k8s之DaemonSet k8s之Job和CronJob k8s之ConfigMap和Secret k8s之Service k8s之Pod k8s之Deployment 容器的本质 docker容器 python迭代器与生成器 python元编程 python垃圾回收机制 python上下文管理器 django rest_framework使用jwt django rest_framework异常处理 django rest_framework 自定义文档 django压缩文件下载 django rest_framework使用pytest单元测试 django restframework choice 自定义输出数据 django Filtering 使用 django viewset 和 Router 配合使用时报的错 django model的序列化 django中使用AbStractUser django.core.exceptions.ImproperlyConfigured Application labels aren't unique, duplicates users django 中 media配置 django 外键引用自身和on_delete参数 django 警告 while time zone support is active Flask使用flask_socketio实现websocket flask结合mongo tornado 文件上传 tornado 使用jwt完成用户异步认证 tornado 用户密码 bcrypt加密 tornado 结合wtforms使用表单操作 tornado finish和write区别 tornado 使用peewee-async 完成异步orm数据库操作使用hue创建ozzie的pyspark action workflow count的性能优化 django rest_framework Authentication django celery 结合使用网站

zhengwenfeng · 2022-08-10 · via 郑文峰的博客

# 简介

并不是真正的实时处理框架，只是按照时间进行微批处理进行，时间可以设置的尽可能的小。

将不同的额数据源的数据经过SparkStreaming 处理之后将结果输出到外部文件系统

特点

低延时能从错误中搞笑的恢复: fault-tolerant 能够运行在成百上千的节点能够将批处理、机器学习、图计算等自框架和Spark Streaming 综合起来使用

粗粒度

Spark Streaming接收到实时数据流，把数据按照指定的时间段切成一片片小的数据块，然后把小的数据块传给Spark Engine处理。

细粒度
数据源 kafka提供了两种数据源。

基础数据源，可以直接通过streamingContext API实现。如文件系统和socket连接
高级的数据源，如Kafka, Flume, Kinesis等等. 可以通过额外的类库去实现。

# 基础数据源

使用官方的案例

/spark/examples/src/main/python/streaming

nc -lk 6789

处理socket数据

示例代码如下: 读取socket中的数据进行流处理

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# local 必须设为2
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)

lines = ssc.socketTextStream("localhost", 9999)

words = lines.flatMap(lambda line: line.split(" "))

pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)

wordCounts.pprint()

ssc.start()
ssc.awaitTermination()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

测试

nc -lk 9999

处理文件系统数据

文件系统(fileStream(that is, HDFSM S3, NFS))暂不支持python，python仅支持文本文件(textFileStream)

示例如下，但未成功，找不到该文件。


lines = ssc.textFileStream("hdfs://txz-data0:9820/user/jim/workflow/crash/python/crash_2_hdfs.py")

1
2
3

streaming context
DStreams

持续化的数据流对DStream操作算子，比如map/flatMap,其实底层会被翻译为对DStream中的每个RDD都做相同的操作，因为一个DStream是由不同批次的RDD所

Input DStreams and Receivers

# 高级数据源

# Spark Streaming 和 kafka 整合

两种模式

receiver 模式

from pyspark.streaming.kafka import KafkaUtils
from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext("local[2]", "NetworkWordCount")
sc.setLogLevel("OFF")
ssc = StreamingContext(sc, 1)

# 创建Kafka streaming
line = KafkaUtils.createStream(ssc, "192.168.0.208:2181", 'test', {"jim_test": 1})

# 分词
words = line.flatMap(lambda line: line.split(" "))
pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)
wordCounts.pprint()

ssc.start()
ssc.awaitTermination()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

no receiver

根据上面的代码替换掉createStream即可。

line = KafkaUtils.createDirectStream(ssc, ["jim_test"], {"metadata.broker.list": "192.168.0.208:9092"})

运行:

spark-submit --jars spark-streaming-kafka-0-8-assembly_2.11-2.4.0.jar test_spark_stream.py

需要下载相应的jar包.下载地址如下，搜索。 https://search.maven.org

jar版本会在运行程序时报错提醒。

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

郑文峰的博客

# 简介

# 基础数据源

# 高级数据源

# Spark Streaming 和 kafka 整合