spark structured stream的Append模式例子

本例子实现一个从源kafka消费消息进行分组聚合后重新输出到目的kafka的例子，参见代码：

from pyspark import SparkConf
from pyspark.sql import SparkSession
import traceback
# import builtins as py_builtin
from pyspark.sql.functions import max
from pyspark.sql.functions import desc
from pyspark.sql.types import StructField, StructType, StringType, LongType
from pyspark.sql.types import *
from pyspark.sql.functions import col, column, expr
from pyspark.sql.functions import *
from pyspark.sql import Rowappname = "test"  # 任务名称
master = "local[*]"  # 单机模式设置
'''
local: 所有计算都运行在一个线程当中，没有任何并行计算，通常我们在本机执行一些测试代码，或者练手，就用这种模式。
local[K]: 指定使用几个线程来运行计算，比如local[4]就是运行4个worker线程。通常我们的cpu有几个core，就指定几个线程，最大化利用cpu的计算能力
local[*]: 这种模式直接帮你按照cpu最多cores来设置线程数了。
'''
# spark_driver_host = "10.0.0.248"try:conf = SparkConf().setAppName(appname). \set('spark.jars.packages', 'org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0') \.set("spark.jars.repositories", 'http://maven.aliyun.com/nexus/content/groups/public/') \.setMaster(master)  # 本地spark = SparkSession.builder.config(conf=conf).getOrCreate()df = spark \.readStream \.format("kafka") \.option("kafka.bootstrap.servers", "localhost:9092") \.option("startingOffsets", "latest") \.option("subscribe", "mykafkatest") \.load()words = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "offset", "timestamp")schema = StructType() \.add("topic", StringType()) \.add("age", StringType()) \\# 通过from_json，定义schema来解析json# res = words.where(instr("value", '{') ==0).select(from_json("value", schema).alias("data")).select("data.*")streamSource = words.where(instr("value", 'topic') > 0).select(from_json("value", schema).alias("data"),col("timestamp")) \.select("data.*", 'timestamp')streamSource = streamSource.select(col('age').cast('int').alias('age'), col('topic'), col('timestamp'))windowedCounts = streamSource.withWatermark("timestamp", "10 seconds") \.groupBy(window(col("timestamp"), '10 seconds', '10 seconds'), col("topic")).count()# query = windowedCounts \#     .writeStream \#     .outputMode('complete') \#     .format('console') \#     .option('truncate', 'false') \#     .start()res = windowedCounts.withColumn('constfield', lit('1'))query = res.select(to_json(struct("topic", "window")).cast('string').alias("key"),to_json(struct("topic", "window", "count")).cast('string').alias("value"))\.writeStream \.format("kafka") \.option("kafka.bootstrap.servers", "localhost:9092") \.option("topic", "mykafkatestdest") \.option("checkpointLocation", '''D:\spark\spark\spark-2.3.0-bin-hadoop2.7\checkpoint\kafkatest2''') \.trigger(processingTime='3 seconds') \.start()query.awaitTermination()print('计算成功！')
except:traceback.print_exc()  # 返回出错信息print('连接出错！')

运行结果：

结论：写入kafka sink时使用的是默认的append输出模式，也就是窗口的信息输出是等到对应的wartermark达到时才会输出的，此时也会顺便清理state，防止state过大。不过在这个例子中也发现一个问题：最后的几条消息一直在等待wartermark的到来，现象就是最后几条消息一直没有聚合输出，不论等待了多少个trigger触发时间，除非再次输入新的几条消息才输出，然而这样新的这几条消息又会一直再次等待wartermark的到达，—这里感觉有点问题，如果只是最后一条消息等待wartermark这能理解，因为append输出模式的语义来说未确定wartermark之前确实不能输出，但是有几条小子都在等待wartermark这里就很难理解了，难道是不同的执行器还有自己的wartermark？需要每个执行器都等到了自己的wartermark才触发？

spark structured stream的Append模式例子相关推荐

spark structured stream的Update模式
spark的update模式的定义为:自动上一次trigger以来有变化的key都会输出到kafka sink中. 下面的例子完整实现一个从kafka消费并聚合消息,然后把聚合消息写入到目标kafka ...
Spark Structured SQL报错：Stream stream joins without equality predicate is not supported
1.背景写一个Spark Structured SQL 任务,任务的功能是对kafka的两个topic进行join处理. select q.sysdt, q.systm, q.event_time ...
Spark Structured Steaming实战
Spark Structured Steaming Spark Structured Streaming 简介什么是 Spark Structured Streaming Structured St ...
【Spark Streaming】（四）基于 Spark Structured Streaming 的开发与数据处理
文章目录一.前言二.Spark Streaming vs Structured Streaming 2.1 Spark Streaming 2.2 Structured Streaming 2.3 ...
kafka 的structured stream 总结
比较重要的几个概念: 1.trigger触发时间,这个触发时间是指每次触发从kafka读取数据的时间间隔,如果不设置,就是尽可能快的意思,上一批处理完马上下一批,如果偶尔停机,而kafka中积累了大量 ...
大数据之Spark:Structured Streaming
目录 1. API 2. 核心思想 3. 应用场景 4.Structured Streaming 实战 1) 读取 Socket 数据 2) 读取目录下文本数据 3) 计算操作 4) 输出在 2.0 ...
Spark Structured Streaming概述
Spark Structured Streaming概述结构化流(Structured Streaming)是基于Spark SQL引擎的流处理引擎,它具有可扩展和容错性.可以使用类似批数据处理的表 ...
Spark2.3（三十五）Spark Structured Streaming源代码剖析(从CSDN和Github中看到别人分析的源代码的文章值得收藏)...
从CSDN中读取到关于spark structured streaming源代码分析不错的几篇文章 spark源码分析--事件总线LiveListenerBus spark事件总线的核心是LiveLi ...
kafka spark Structured streaming整合后集群报错KafkaConsumer.subscribe(Ljava/util/Collection;)V
简介整个项目架构是在CDH中,,然后spark Structured streaming消费kafka. spark 2.3版本 kafka0.10版本 <!-- spark sql kafk ...

spark structured stream的Append模式例子

spark structured stream的Append模式例子相关推荐

最新文章

热门文章