大数据平台架构技术详解

# 大数据平台架构核心代码示例 import pyspark from pyspark.sql import SparkSession from pyspark.sql.functions import * from pyspark.sql.types import * import kafka from kafka import KafkaProducer, KafkaConsumer import json import time from datetime import datetime, timedelta class BigDataPlatform: def __init__(self, config): self.config = config self.spark = None self.kafka_producer = None self.kafka_consumer = None def initialize_spark(self): """初始化Spark会话""" self.spark = SparkSession.builder \ .appName("BigDataPlatform") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \ .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \ .getOrCreate() # 设置日志级别 self.spark.sparkContext.setLogLevel("WARN") print("Spark会话初始化完成") def initialize_kafka(self): """初始化Kafka连接""" self.kafka_producer = KafkaProducer( bootstrap_servers=self.config['kafka_servers'], value_serializer=lambda v: json.dumps(v).encode('utf-8') ) self.kafka_consumer = KafkaConsumer( self.config['kafka_topic'], bootstrap_servers=self.config['kafka_servers'], value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) print("Kafka连接初始化完成") def batch_processing(self, data_path): """批处理数据处理""" print(f"开始批处理数据: {data_path}") # 读取数据 df = self.spark.read \ .option("header", "true") \ .option("inferSchema", "true") \ .csv(data_path) # 数据清洗和转换 cleaned_df = df.filter(col("id").isNotNull()) \ .withColumn("created_at", to_timestamp(col("created_at"))) \ .withColumn("processed_at", current_timestamp()) # 数据聚合分析 aggregated_df = cleaned_df.groupBy("category") \ .agg( count("*").alias("total_count"), sum("amount").alias("total_amount"), avg("amount").alias("avg_amount"), max("amount").alias("max_amount"), min("amount").alias("min_amount") ) # 保存结果 output_path = f"{self.config['output_path']}/batch_processing_{datetime.now().strftime('%Y%m%d_%H%M%S')}" aggregated_df.write \ .mode("overwrite") \ .parquet(output_path) print(f"批处理完成，结果保存到: {output_path}") return aggregated_df def stream_processing(self): """流处理数据处理""" print("开始流处理数据") # 创建流式DataFrame stream_df = self.spark \ .readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", self.config['kafka_servers']) \ .option("subscribe", self.config['kafka_topic']) \ .option("startingOffsets", "latest") \ .load() # 解析JSON数据 parsed_df = stream_df.select( from_json(col("value").cast("string"), self.get_schema()).alias("data") ).select("data.*") # 实时数据处理 processed_df = parsed_df \ .withColumn("processed_at", current_timestamp()) \ .withColumn("hour", hour(col("timestamp"))) \ .withColumn("day", dayofmonth(col("timestamp"))) # 实时聚合 aggregated_stream = processed_df.groupBy( window(col("timestamp"), "1 minute"), col("category") ).agg( count("*").alias("count"), sum("amount").alias("total_amount") ) # 输出到控制台（实际项目中可输出到数据库或其他存储） query = aggregated_stream.writeStream \ .outputMode("complete") \ .format("console") \ .option("truncate", False) \ .trigger(processingTime="10 seconds") \ .start() return query def get_schema(self): """定义数据Schema""" return StructType([ StructField("id", StringType(), True), StructField("category", StringType(), True), StructField("amount", DoubleType(), True), StructField("timestamp", TimestampType(), True), StructField("user_id", StringType(), True) ]) def data_quality_check(self, df): """数据质量检查""" print("开始数据质量检查") # 检查数据完整性 total_count = df.count() null_count = df.filter(col("id").isNull()).count() completeness = (total_count - null_count) / total_count * 100 # 检查数据准确性 valid_amount_count = df.filter(col("amount") > 0).count() accuracy = valid_amount_count / total_count * 100 # 检查数据一致性 duplicate_count = df.count() - df.dropDuplicates().count() consistency = (total_count - duplicate_count) / total_count * 100 quality_report = { "total_records": total_count, "completeness": round(completeness, 2), "accuracy": round(accuracy, 2), "consistency": round(consistency, 2), "quality_score": round((completeness + accuracy + consistency) / 3, 2) } print(f"数据质量报告: {quality_report}") return quality_report def send_to_kafka(self, data): """发送数据到Kafka""" try: self.kafka_producer.send(self.config['kafka_topic'], data) self.kafka_producer.flush() print(f"数据发送成功: {data['id']}") except Exception as e: print(f"数据发送失败: {e}") def consume_from_kafka(self): """从Kafka消费数据""" print("开始消费Kafka数据") for message in self.kafka_consumer: try: data = message.value print(f"接收到数据: {data}") # 处理数据 processed_data = self.process_realtime_data(data) # 存储到数据库或进行其他处理 self.store_data(processed_data) except Exception as e: print(f"数据处理失败: {e}") def process_realtime_data(self, data): """实时数据处理""" # 添加处理时间戳 data['processed_at'] = datetime.now().isoformat() # 数据验证和清洗 if 'amount' in data and data['amount'] < 0: data['amount'] = 0 return data def store_data(self, data): """存储数据""" # 这里可以实现数据存储逻辑 # 例如存储到数据库、文件系统等 print(f"数据存储: {data['id']}") def create_data_lake_table(self, table_name, data_path): """创建数据湖表""" print(f"创建数据湖表: {table_name}") # 创建Delta表 df = self.spark.read.parquet(data_path) df.write \ .format("delta") \ .mode("overwrite") \ .option("path", f"{self.config['delta_path']}/{table_name}") \ .saveAsTable(table_name) print(f"数据湖表创建完成: {table_name}") def optimize_query_performance(self, query): """查询性能优化""" print("开始查询性能优化") # 启用自适应查询执行 self.spark.conf.set("spark.sql.adaptive.enabled", "true") self.spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") # 启用列式存储 self.spark.conf.set("spark.sql.parquet.enableVectorizedReader", "true") # 启用谓词下推 self.spark.conf.set("spark.sql.parquet.filterPushdown", "true") # 执行查询 start_time = time.time() result = self.spark.sql(query) result.show() end_time = time.time() print(f"查询执行时间: {end_time - start_time:.2f}秒") return result def monitor_system_health(self): """系统健康监控""" print("开始系统健康监控") # 检查Spark应用状态 spark_status = self.spark.sparkContext.statusTracker() print(f"Spark应用状态: {spark_status.getExecutorInfos()}") # 检查存储空间 # 这里可以实现存储空间检查逻辑 # 检查数据质量 # 这里可以实现数据质量监控逻辑 health_report = { "timestamp": datetime.now().isoformat(), "spark_status": "healthy", "storage_status": "healthy", "data_quality": "good" } print(f"系统健康报告: {health_report}") return health_report # 使用示例 def main(): # 配置参数 config = { 'kafka_servers': 'localhost:9092', 'kafka_topic': 'bigdata_topic', 'output_path': '/data/output', 'delta_path': '/data/delta' } # 创建大数据平台实例 platform = BigDataPlatform(config) # 初始化组件 platform.initialize_spark() platform.initialize_kafka() # 批处理示例 batch_result = platform.batch_processing('/data/input/sample_data.csv') # 数据质量检查 quality_report = platform.data_quality_check(batch_result) # 创建数据湖表 platform.create_data_lake_table('sample_table', '/data/output') # 查询性能优化 query = "SELECT category, sum(amount) as total FROM sample_table GROUP BY category" platform.optimize_query_performance(query) # 系统健康监控 platform.monitor_system_health() print("大数据平台架构系统运行完成") if __name__ == "__main__": main()