2025年Python实时数据处理专项训练试卷

2025年Python实时数据处理专项训练试卷考试时间：______分钟总分：______分姓名：______一、简述流式数据处理与批量数据处理在处理逻辑、性能特点、适用场景等方面的主要区别。二、假设你正在使用`kafka-python`库从Kafka主题`sensor_data`读取实时温度传感器数据流，数据以JSON格式发送，每个JSON对象包含`timestamp`（时间戳，字符串格式如`"2023-10-2710:00:00"`）和`temperature`（温度值，浮点数）两个字段。请编写Python代码片段，实现以下功能：1.连接到Kafka集群（假设brokers地址为`'localhost:9092'`）。2.从`sensor_data`主题读取数据流。3.解析每个JSON消息，提取`temperature`值。4.如果`temperature`值有效（非空且为浮点数），则将其添加到名为`temperatures`的列表中。5.每读取10条有效数据，计算并打印这10条数据的平均温度。三、使用Pandas库，假设你有一个名为`df`的DataFrame，其中包含以下列：`timestamp`（时间戳，PandasTimestamp类型）、`sensor_id`（传感器ID，字符串）、`reading`（读数，浮点数）。请编写代码片段，实现以下数据处理任务：1.将`df`按照`timestamp`列进行升序排序。2.对`df`应用滑动窗口（SlidingWindow），窗口大小为5行。对于窗口内的每一行数据，计算其`reading`值与窗口内前一行的`reading`值的变化率（当前行读数-前一行读数），并将此变化率作为新列`change_rate`添加到DataFrame中。注意处理窗口起始行的`change_rate`计算。3.筛选出`sensor_id`为`'sensor_A'`且`change_rate`绝对值大于0.5的记录，并将这些记录存储到新的DataFrame`dfAlerts`中。四、假设你需要使用Dask进行分布式处理，处理一个包含数百万行传感器数据的CSV文件`sensor_data.csv`（包含`timestamp`,`sensor_id`,`temperature`等列）。请简述使用DaskDataFrame进行以下操作的代码思路：1.读取`sensor_data.csv`文件。2.对所有数据计算每分钟内的平均温度。3.筛选出`temperature`超过某个阈值（例如35.0）的记录。4.将筛选出的记录按`sensor_id`分组，并计算每个传感器的记录数量。五、在PySpark环境中，使用SparkStreaming处理来自Kafka的实时传感器数据（与第二题描述类似）。请简述如何实现以下功能：1.创建一个StreamingContext，设置适当的微批处理间隔（例如5秒）。2.读取Kafka主题`sensor_data`的数据流。3.对每个微批处理的数据，使用窗口函数计算过去10分钟内每个`sensor_id`的平均`temperature`。4.将计算结果实时写入一个HDFS路径（例如`/user/hadoop/streaming_results/`）。六、考虑一个实时数据处理系统，需要处理来自多个传感器的数据流，并计算全局的平均温度。请简述在以下两种情况下，系统如何处理数据乱序（Out-of-OrderData）问题，并解释水位线（Watermark）的概念及其作用：1.使用PySparkStreaming处理数据。2.使用FlinkStreaming处理数据。七、简述Exactly-Once(EOS)、At-Least-Once(ALO)和At-Most-Once(AMO)这三种流式数据处理语义的区别。在哪些场景下，保证EOS语义通常是必须的？请举例说明。八、假设实时处理任务完成后，需要将结果数据存储到不同的系统以供后续分析或应用使用。请比较以下几种存储系统的特点，并说明它们分别适用于实时处理结果的哪种场景：1.Redis2.InfluxDB3.ApacheKafka4.关系型数据库（如PostgreSQL）九、请描述在Python实时数据处理任务中，使用异步编程（例如`asyncio`库）可能带来的好处，并给出一个可能适合使用异步IO处理实时数据的场景示例。试卷答案一、流式数据处理是连续、近乎实时地处理数据流，强调低延迟和事件驱动；批量处理是周期性地收集一批数据后进行集中处理，延迟较高。流式处理适用于需要快速响应的场景（如实时监控、欺诈检测），而批量处理适用于对数据完整性要求高、计算资源有限或可以接受一定延迟的场景（如大规模报表生成、离线分析）。流式处理通常需要处理状态管理、乱序数据等问题，而批量处理相对简单。二、```pythonfromkafkaimportKafkaConsumerimportjsonconsumer=KafkaConsumer('sensor_data',bootstrap_servers='localhost:9092',auto_offset_reset='earliest',value_deserializer=lambdax:json.loads(x.decode('utf-8')))temperatures=[]count=0formessageinconsumer:try:temp=message.value.get('temperature')iftempisnotNoneandisinstance(temp,float):temperatures.append(temp)count+=1ifcount==10:avg_temp=sum(temperatures)/len(temperatures)print(f"Averagetemperatureforlast10readings:{avg_temp}")temperatures=[]#Resetfornextwindowcount=0exceptExceptionase:print(f"Errorprocessingmessage:{e}")```解析思路：首先创建KafkaConsumer实例连接到集群并指定主题和反序列化方式。使用for循环迭代消费消息，尝试解析JSON并提取温度值。若温度值有效，则添加到列表并计数。每当积累10条有效数据时，计算平均值并打印，然后重置列表和计数器以处理下一个10条数据的窗口。三、```pythonimportpandasaspd#假设df已经定义并包含所需列#1.按timestamp排序df=df.sort_values(by='timestamp')#2.计算变化率df['change_rate']=df['reading'].diff()#diff()计算与前一行值的差#3.筛选sensor_A且change_rate绝对值大于0.5的记录dfAlerts=df[(df['sensor_id']=='sensor_A')&(abs(df['change_rate'])>0.5)]```解析思路：使用`sort_values`对DataFrame按`timestamp`列进行排序。使用`diff()`函数计算`reading`列与前一行值的差，即变化率，并将结果存储在新列`change_rate`中。`diff()`函数自动处理起始行的变化率（通常显示为NaN）。最后，使用布尔索引筛选出满足`sensor_id`为`'sensor_A'`且`change_rate`绝对值大于0.5的记录。四、代码思路：```pythonimportdask.dataframeasdd#1.读取CSV文件ddf=dd.read_csv('sensor_data.csv')#2.计算每分钟内的平均温度#需要确保timestamp列是datetime类型，并设置为索引ddf['timestamp']=dd.to_datetime(ddf['timestamp'])ddf=ddf.set_index('timestamp')mean_temp_per_minute=ddf['temperature'].resample('T').mean()#3.筛选temperature超过35.0的记录high_temp_records=ddf[ddf['temperature']>35.0]#4.计算每个传感器的记录数量sensor_counts=ddf['sensor_id'].value_counts()```解析思路：使用`dd.read_csv`读取CSV文件为DaskDataFrame。将`timestamp`列转换为Pandasdatetime类型并设置为索引，这是进行时间序列操作的关键。使用`resample('T')`方法对`temperature`列按分钟进行重采样，并调用`mean()`计算每分钟的平均温度。使用布尔索引筛选出`temperature`大于35.0的记录。使用`value_counts()`对`sensor_id`列进行计数，得到每个传感器的记录数量。Dask会自动进行分布式计算。五、解析思路：1.创建`SparkSession`，然后使用`SparkSession`的`streamingContext`方法创建`StreamingContext`，设置合适的`batchDuration`（例如5秒）。2.使用`StreamingContext`的`kafka`消费组创建读取Kafka数据的`DataFrame`，指定主题、Kafka服务器、消费组ID、反序列化方式等。3.将Kafka`DataFrame`转换为`Dataset`，应用窗口函数（例如`groupByWindow`结合`mean`），指定窗口大小（例如10分钟）和窗口滑动步长，计算每个窗口内每个`sensor_id`的平均`temperature`。4.使用`writeStream`将结果`DataFrame`或`Dataset`的输出实时写入HDFS路径。配置写入参数（如`checkpointLocation`），并调用`start()`和`awaitTermination()`启动流处理作业。六、1.PySparkStreaming处理数据乱序：需要在创建`DataFrameReader`时指定`watermarkInterval`（水位线间隔），例如`watermarkInterval="10minutes"`。PySpark会根据时间戳计算水位线，并丢弃超过水位线的数据。可以在窗口函数计算时使用`ignoreWatermark`参数或通过过滤条件`df.filter(df.timestamp<=watermark)`来处理乱序数据。2.FlinkStreaming处理数据乱序：Flink内置了强大的事件时间和水位线处理机制。需要在源函数（SourceFunction）中实现`extractTimestamp`方法返回事件时间戳，并设置`WatermarkStrategy`指定水位线生成逻辑（例如`withTimestampAssigner`和`withWatermark`）。Flink会根据水位线自动处理乱序事件，确保状态更新的一致性。可以在窗口操作或ProcessFunction中显式地使用水位线进行状态更新和结果计算。七、EOS保证每个事件只被处理一次，适用于金融交易、订单处理等对数据一致性要求极高的场景。ALO保证每个事件至少被处理一次，可能存在重复处理，适用于对一致性要求不是特别严格的场景，如日志聚合。AMO保证每个事件最多被处理一次，不保证处理顺序，适用于对吞吐量要求极高、可以容忍少量重复处理的场景，如简单的计数统计。EOS语义通常在需要精确业务结果且不允许错漏的场景下必须保证，例如支付系统。八、1.Redis：内存数据库，读写速度快，支持多种数据结构（字符串、哈希、列表、集合等）。适用于需要极低延迟访问、缓存、实时排行榜、计数器等场景。2.InfluxDB：时序数据库，专为时间序列数据设计，优化了时间序列数据的存储和查询（使用InfluxQL/Flux）。适用于监控、物联网、传感器数据等需要高效存储和查询时间序列数据的场景。3.ApacheKafka：分布式流处理平台，高吞吐量，可扩展性强，用于构建实时数据管道和流应用。适用于需要大规模处理高吞吐量数据流、解耦系统、数据集成等场景。4.关系型数据库（如PostgreSQL）：结构化数据存储，支持SQL查询