StructuredStreaming 的介绍

发布时间：2025-06-24 19:20:59  作者：北方职教升学中心  阅读量：271

新开发的流式引擎致力于为批处理和流处理提供统一的高性能API。Continuous Processing毫秒延迟(2.3).0)。

import osfrom pyspark.sql import SparkSessionfrom pyspark.sql.functions import explodefrom pyspark.sql.functions import splitif __name__ == '__main__':    # 配置环境    os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_241'    # 配置Hadoop路径，StructuredStreaming 的介绍。
Structured Streaming Programming Guide - Spark 3.5.3 Documentation。
四、
一、允许用户简单地编写高性能的流程处理程序Structured，
Spark在2016年Spark2.0版本中发布了新的流计算API:Structured streaming结构化流。比如Event Time(事件时间)支持，
1.基于微批，
二、同时也考虑了和Spark 更好地集成其它组件。就像编写批处理程序一样 streaming不是Spark Streaming的简单改进，Source。
3.流量处理的API应用层不统一(流量DStream-底层为RDD，同时也考虑了和Spark 更好地集成其他组件。
三、如果有相似之处，以及Spark社区和Databricks众多客户的反馈，Structured streaming是一种基于sparkSOL引擎的可扩展性和容错性的新型流处理引擎。
structured Streaming统一了流程和批次的编程模型，
4.不支持Eventtime事件的时间。
案例一：Socket。SparkStreaming 的不足。编程模型。Stream-Streamjoin(2.3.0)，
用pyspark编写StructStreaming的入门案例，
Structured 新引擎Streaming也实现了SparkStreaming之前没有的一些功能，所有代码都可以亲测。在SparkSQL和Sparkstreaming开发过程中吸取了教训，延迟高不能实现真正的实时。纯粹是巧合，
Structured Streaming Programming Guide - Spark 3.5.3 Documentation。
5.数据Exactly-Once(只是一个语义)需要手动实现。前解压的路径是前解压的路径    os.environ['HADOOP_HOME'] = 'D:/hadoop-3.3.1'    # 配备base环境Python分析器的路径    os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配备base环境Python分析器的路径    os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'    os.environ['HADOOP_USER_NAME'] = 'root'    spark = SparkSession \        .builder \        .appName("StructuredNetworkWordCount") \        .getOrCreate()    # Create DataFrame representing the stream of input lines from connection to localhost:9999    lines = spark \        .readStream \        .format("socket") \        .option("host", “bigdata01” \        .option("port", 9999) \        .load()    # Split the lines into words    words = lines.select(       explode(           split(lines.value, " ")       ).alias("word")    )    # Generate running word count    wordCounts = words.groupBy("word").count()    # Start running the query that prints the running counts to the console    query = wordCounts \        .writeStream \        .outputMode("complete") \        .format("console") \        .start()    query.awaitTermination()    spa。批量DF/DS/RDD)。
2.基于RDD的DStream不直接支持SQL。