大数据-玩转数据-Spark-Structured Streaming 容错（python版）

本文主要是介绍大数据-玩转数据-Spark-Structured Streaming 容错（python版），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

大数据-玩转数据-Spark-Structured Streaming 容错（python版）

说明：
由于网络问题，链路中断，系统崩溃，JVM故障都会导致数据流的运行结果出现错误，Spark设计了输入源，执行引擎和接收器多个松散耦合组件隔离故障。

输入源通过位置偏移量来记录目前所处位置，引擎通过检查点保存中间状态，接收器使用“幂等”的接收器来保障输出的稳定性。

我们希望数据是它产生的时间，而不是到达的时间，Spark模型当中，事件时间是数据中的一列，为了避免存储空间无限扩大，同时还引入“水印”机制，将超过时间阈值的数据抛弃掉。

代码举例

#!/usr/bin/env python3
# -*- coding: utf-8 -*-# 导入需要用到的模块
import os
import shutil
from functools import partialfrom pyspark.sql import SparkSession
from pyspark.sql.functions import window
from pyspark.sql.types import StructType, StructField
from pyspark.sql.types import TimestampType, StringType# 定义CSV文件的路径常量
TEST_DATA_DIR = '/tmp/testdata/'
TEST_DATA_DIR_SPARK = 'file:///tmp/testdata/'# 测试的环境搭建，判断CSV文件夹是否存在，如果存在则删除旧数据，并建立文件夹
def test_setUp():if os.path.exists(TEST_DATA_DIR):shutil.rmtree(TEST_DATA_DIR, ignore_errors=True)os.mkdir(TEST_DATA_DIR)# 测试环境的恢复，对CSV文件夹进行清理
def test_tearDown():if os.path.exists(TEST_DATA_DIR):shutil.rmtree(TEST_DATA_DIR, ignore_errors=True)# 写模拟输入的函数，传入CSV文件名和数据。注意写入应当是原子性的
# 如果写入时间较长，应当先写到临时文件在移动到CSV目录内。
# 这里采取直接写入的方式。
def write_to_cvs(filename, data):with open(TEST_DATA_DIR + filename, "wt", encoding="utf-8") as f:f.write(data)if __name__ == "__main__":test_setUp()# 定义模式，为字符串类型的word和时间戳类型的eventTime两个列组成schema = StructType([StructField("word", StringType(), True),StructField("eventTime", TimestampType(), True)])spark = SparkSession \.builder \.appName("StructuredNetworkWordCountWindowedDelay") \.getOrCreate()spark.sparkContext.setLogLevel('WARN')lines = spark \.readStream \.format('csv') \.schema(schema) \.option("sep", ";") \.option("header", "false") \.load(TEST_DATA_DIR_SPARK)# 定义窗口windowDuration = '1 hour'windowedCounts = lines \.withWatermark("eventTime", "1 hour") \.groupBy('word', window('eventTime', windowDuration)) \.count()query = windowedCounts \.writeStream \.outputMode("update") \.format("console") \.option('truncate', 'false') \.trigger(processingTime="8 seconds") \.start()# 写入测试文件file1.cvswrite_to_cvs('file1.cvs', """
正常;2018-10-01 08:00:00
正常;2018-10-01 08:10:00
正常;2018-10-01 08:20:00
""")# 处理当前数据query.processAllAvailable()# 这时候事件时间更新到上次看到的最大的2018-10-01 08:20:00write_to_cvs('file2.cvs', """
正常;2018-10-01 20:00:00
一小时以内延迟到达;2018-10-01 10:00:00
一小时以内延迟到达;2018-10-01 10:50:00
""")# 处理当前数据query.processAllAvailable()# 这时候事件时间更新到上次看到的最大的2018-10-01 20:00:00write_to_cvs('file3.cvs', """
正常;2018-10-01 20:00:00
一小时外延迟到达;2018-10-01 10:00:00
一小时外延迟到达;2018-10-01 10:50:00
一小时以内延迟到达;2018-10-01 19:00:00
""")# 处理当前数据query.processAllAvailable()query.stop()test_tearDown()

这篇关于大数据-玩转数据-Spark-Structured Streaming 容错（python版）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---