# 導(dǎo)入Kafka生產(chǎn)者模塊
from confluent_kafka import Producer
# 導(dǎo)入JSON處理模塊（原代碼缺失此導(dǎo)入）
import json

# 配置Kafka集群連接參數(shù)（多個(gè)broker用逗號(hào)分隔）
conf = {'bootstrap.servers': 'kafka1:9092,kafka2:9092'}

# 創(chuàng)建Kafka生產(chǎn)者實(shí)例
producer = Producer(conf)

# 定義消息投遞結(jié)果回調(diào)函數(shù)
def delivery_report(err, msg):
    """處理消息發(fā)送后的回調(diào)結(jié)果"""
    # 如果發(fā)送失敗則打印錯(cuò)誤
    if err is not None:
        print(f'Message delivery failed: {err}')
    # 發(fā)送成功時(shí)打印消息元數(shù)據(jù)
    else:
        print(f'Message delivered to: '
              f'topic={msg.topic()} '
              f'partition={msg.partition()} '
              f'offset={msg.offset()}')

# 構(gòu)造用戶事件數(shù)據(jù)（Python字典格式）
user_event = {
    'user_id': 101,
    'action': 'payment',
    'amount': 299.9
}

# 使用生產(chǎn)者發(fā)送消息到指定主題
producer.produce(
    topic='user_events',  # 目標(biāo)Kafka主題名稱
    key=str(user_event['user_id']),  # 設(shè)置消息鍵（按用戶ID分區(qū)）
    value=json.dumps(user_event),  # 將字典轉(zhuǎn)為JSON字符串
    callback=delivery_report  # 指定投遞結(jié)果回調(diào)函數(shù)
)

# 可選：在發(fā)送后立即輪詢事件隊(duì)列（處理回調(diào)）
# 確保在flush前處理已發(fā)送消息的回調(diào)
producer.poll(0)

# 強(qiáng)制刷新生產(chǎn)者緩沖區(qū)，確保所有消息完成傳輸
# 會(huì)阻塞直到所有消息得到broker確認(rèn)或超時(shí)
producer.flush()

# 注意：實(shí)際生產(chǎn)環(huán)境通常不會(huì)每條消息都flush
# 可考慮批量發(fā)送或定時(shí)刷新以提高吞吐量

# 導(dǎo)入必要的PySpark模塊
from pyspark import SparkConf, SparkContext

# 初始化Spark配置（實(shí)際應(yīng)用中可配置集群參數(shù)）
conf = SparkConf().setAppName("WordCount")  # 設(shè)置應(yīng)用名稱
sc = SparkContext(conf=conf)  # 創(chuàng)建SparkContext實(shí)例

# 從HDFS分布式文件系統(tǒng)加載文本數(shù)據(jù)創(chuàng)建初始RDD
# 參數(shù)：HDFS文件路徑（假設(shè)為訪問日志）
text_rdd = sc.textFile("hdfs://logs/access.log")  # 返回RDD[String]類型

# ===== 轉(zhuǎn)換操作（Transformations）=====
# 惰性操作：僅定義計(jì)算邏輯，不立即執(zhí)行

# 扁平化操作：將每行文本分割成單詞
# flatMap: 每行輸入 -> 多個(gè)輸出元素（單詞）
words_rdd = text_rdd.flatMap(lambda line: line.split(" "))

# 映射操作：將每個(gè)單詞轉(zhuǎn)換為(單詞, 1)鍵值對(duì)
# map: 每個(gè)單詞 -> (word, 1) 二元組
pairs_rdd = words_rdd.map(lambda word: (word, 1))

# 按鍵聚合：對(duì)相同單詞的計(jì)數(shù)值進(jìn)行累加
# reduceByKey: 對(duì)相同key的值執(zhí)行聚合函數(shù)（這里是加法）
counts_rdd = pairs_rdd.reduceByKey(lambda a, b: a + b)

# ===== 行動(dòng)操作（Action）=====
# 觸發(fā)實(shí)際計(jì)算并返回結(jié)果到驅(qū)動(dòng)程序

# 獲取詞頻最高的10個(gè)單詞（按詞頻降序）
# takeOrdered: 返回按指定鍵排序的前N個(gè)元素
# key=lambda x: -x[1]: 按計(jì)數(shù)值（元組第二項(xiàng)）降序排列
top10 = counts_rdd.takeOrdered(10, key=lambda x: -x[1])

# 打印結(jié)果到控制臺(tái)
print("Top10高頻詞:", top10)

# 可選：關(guān)閉SparkContext釋放資源
# 在長時(shí)間運(yùn)行的Spark應(yīng)用（如Spark Streaming）中可能不立即關(guān)閉
sc.stop()

# 導(dǎo)入必要的PySpark模塊
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType, DoubleType

# 創(chuàng)建SparkSession實(shí)例（流處理程序的入口點(diǎn)）
spark = SparkSession.builder \
    .appName("KafkaPaymentMonitor") \  # 設(shè)置應(yīng)用名稱
    .config("spark.sql.shuffle.partitions", "4") \  # 優(yōu)化小規(guī)模數(shù)據(jù)處理
    .getOrCreate()  # 獲取或創(chuàng)建會(huì)話實(shí)例

# 定義JSON事件的結(jié)構(gòu)化模式
# 對(duì)應(yīng)Kafka消息中的JSON格式：{'user_id':101, 'action':'payment', 'amount':299.9, 'timestamp':'2023-01-01T12:00:00Z'}
event_schema = StructType([
    StructField("user_id", IntegerType(), True),      # 用戶ID整型字段
    StructField("action", StringType(), True),        # 行為類型字符串字段
    StructField("amount", DoubleType(), True),        # 支付金額雙精度字段
    StructField("timestamp", TimestampType(), True)   # 事件時(shí)間戳字段（關(guān)鍵用于窗口計(jì)算）
])

# ===== 定義Kafka流源 =====
# 創(chuàng)建流式DataFrame，從Kafka持續(xù)讀取數(shù)據(jù)
df = spark.readStream \  # 創(chuàng)建流式讀取器
    .format("kafka") \  # 指定Kafka數(shù)據(jù)源格式
    .option("kafka.bootstrap.servers", "kafka1:9092") \  # Kafka集群地址
    .option("subscribe", "user_events") \  # 訂閱的主題名稱
    .option("startingOffsets", "latest") \  # 從最新偏移量開始（可選：earliest）
    .option("failOnDataLoss", "false") \  # 容忍數(shù)據(jù)丟失（生產(chǎn)環(huán)境推薦）
    .load()  # 加載流數(shù)據(jù)源

# ===== 數(shù)據(jù)處理管道 =====
# 步驟1：解析JSON并過濾支付事件
payments = df.select(
    # 解析value字段（二進(jìn)制轉(zhuǎn)為字符串，再按schema解析為結(jié)構(gòu)化數(shù)據(jù)）
    from_json(col("value").cast("string"), event_schema).alias("data"), 
    # 保留Kafka消息自帶的時(shí)間戳（可選，通常使用事件時(shí)間）
    col("timestamp").alias("kafka_timestamp")  
).filter("data.action = 'payment'")  # 過濾出支付事件

# 步驟2：實(shí)時(shí)窗口聚合（每5分鐘窗口按用戶統(tǒng)計(jì)支付次數(shù)）
windowed_count = payments.groupBy(
    # 基于事件時(shí)間創(chuàng)建5分鐘滾動(dòng)窗口
    window(col("data.timestamp"), "5 minutes"),  # 使用事件時(shí)間字段
    col("data.user_id")  # 按用戶ID分組
).count()  # 計(jì)算每個(gè)(窗口,用戶)組合的支付次數(shù)

# ===== 輸出結(jié)果 =====
# 創(chuàng)建流式查詢，將聚合結(jié)果輸出到控制臺(tái)
query = windowed_count.writeStream \
    .outputMode("complete") \  # 完整輸出模式（更新整個(gè)結(jié)果集）
    .format("console") \  # 輸出到控制臺(tái)（生產(chǎn)環(huán)境可用Kafka/文件系統(tǒng)）
    .option("truncate", "false") \  # 顯示完整內(nèi)容（不截?cái)啵?
    .option("numRows", 20) \  # 每次觸發(fā)顯示20行
    .trigger(processingTime="1 minute") \  # 每分鐘觸發(fā)一次計(jì)算
    .start()  # 啟動(dòng)流處理作業(yè)

# 等待查詢終止（實(shí)際應(yīng)用可添加優(yōu)雅停止邏輯）
query.awaitTermination()

# 導(dǎo)入必要的PySpark模塊
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# 創(chuàng)建SparkSession實(shí)例（流處理程序的入口點(diǎn)）
# 使用builder模式配置并創(chuàng)建Spark會(huì)話
spark = SparkSession.builder \
    .appName("KafkaIoTStreamProcessor") \  # 設(shè)置應(yīng)用名稱
    .config("spark.sql.shuffle.partitions", "8") \  # 設(shè)置shuffle分區(qū)數(shù)（根據(jù)集群規(guī)模調(diào)整）
    .config("spark.streaming.backpressure.enabled", "true") \  # 啟用背壓機(jī)制（動(dòng)態(tài)調(diào)整接收速率）
    .getOrCreate()  # 獲取或創(chuàng)建會(huì)話實(shí)例

# 定義IoT設(shè)備遙測(cè)數(shù)據(jù)的結(jié)構(gòu)化模式
# 假設(shè)JSON格式：{"device_id": "sensor-001", "temperature": 23.5, "humidity": 45.2, "timestamp": "2023-01-01T12:00:00Z"}
iot_schema = StructType([
    StructField("device_id", StringType(), True),      # 設(shè)備ID字符串
    StructField("temperature", DoubleType(), True),    # 溫度值（雙精度浮點(diǎn)）
    StructField("humidity", DoubleType(), True),       # 濕度值（雙精度浮點(diǎn)）
    StructField("timestamp", TimestampType(), True)    # 數(shù)據(jù)采集時(shí)間戳
])

# ===== 創(chuàng)建Kafka流源 =====
# 定義從Kafka讀取數(shù)據(jù)的結(jié)構(gòu)化流
stream = spark.readStream \  # 創(chuàng)建流式讀取器
    .format("kafka") \  # 指定Kafka數(shù)據(jù)源格式
    .option("kafka.bootstrap.servers", "kafka1:9092") \  # Kafka集群地址（逗號(hào)分隔多個(gè)broker）
    .option("subscribe", "iot_telemetry") \  # 訂閱的主題名稱（可逗號(hào)分隔多個(gè)主題）
    .option("group.id", "spark-streaming-group") \  # 消費(fèi)者組ID（用于偏移量管理）
    .option("startingOffsets", "earliest") \  # 從最早偏移量開始（可選：latest, 或指定JSON偏移量）
    .option("failOnDataLoss", "false") \  # 容忍數(shù)據(jù)丟失（Kafka主題刪除或偏移量超出范圍時(shí)不失?。?
    .option("maxOffsetsPerTrigger", 10000) \  # 每批處理的最大消息數(shù)（控制批處理大?。?
    .option("kafka.security.protocol", "SASL_SSL") \  # 安全協(xié)議（生產(chǎn)環(huán)境需要）
    .option("kafka.sasl.mechanism", "PLAIN") \  # SASL機(jī)制（生產(chǎn)環(huán)境需要）
    .load()  # 加載流數(shù)據(jù)源

# ===== 數(shù)據(jù)處理管道 =====
# 解析JSON數(shù)據(jù)并轉(zhuǎn)換為結(jié)構(gòu)化格式
parsed_data = stream.select(
    col("key").cast("string").alias("device_key"),  # 可選：轉(zhuǎn)換消息鍵
    from_json(col("value").cast("string"), iot_schema).alias("data"),  # 解析JSON值
    col("topic").alias("kafka_topic"),  # 原始Kafka主題
    col("partition").alias("kafka_partition"),  # Kafka分區(qū)
    col("offset").alias("kafka_offset"),  # 消息偏移量
    col("timestamp").alias("kafka_timestamp")  # Kafka消息時(shí)間戳
).select("device_key", "data.*", "kafka_topic", "kafka_partition", "kafka_offset", "kafka_timestamp")  # 展平嵌套結(jié)構(gòu)

# 過濾異常值（示例：溫度在合理范圍內(nèi)）
filtered_data = parsed_data.filter(
    (col("temperature") >= -40) & 
    (col("temperature") <= 100) &
    (col("humidity") >= 0) & 
    (col("humidity") <= 100)
)

# ===== 輸出結(jié)果 =====
# 創(chuàng)建流式查詢，將處理后的數(shù)據(jù)寫入控制臺(tái)（用于調(diào)試）
# 生產(chǎn)環(huán)境通常會(huì)寫入其他系統(tǒng)（如HDFS、Kafka、數(shù)據(jù)庫等）
query = filtered_data.writeStream \
    .outputMode("append") \  # 追加模式（只輸出新數(shù)據(jù)）
    .format("console") \  # 輸出到控制臺(tái)（開發(fā)/調(diào)試用）
    .option("truncate", "false") \  # 顯示完整內(nèi)容（不截?cái)啵?
    .option("numRows", 100) \  # 每次觸發(fā)顯示100行
    .trigger(processingTime="30 seconds") \  # 每30秒觸發(fā)一次微批處理
    .option("checkpointLocation", "/checkpoints/iot_stream") \  # 檢查點(diǎn)目錄（保證容錯(cuò)性）
    .start()  # 啟動(dòng)流處理作業(yè)

# 等待流查詢終止（通常持續(xù)運(yùn)行直到手動(dòng)停止）
# 實(shí)際應(yīng)用中可添加優(yōu)雅停止邏輯（如響應(yīng)終止信號(hào)）
query.awaitTermination()

# 可選：在程序退出時(shí)停止Spark會(huì)話
spark.stop()

# 導(dǎo)入必要的PySpark模塊
from pyspark.sql import SparkSession
from pyspark.sql.functions import expr, window, count, col
from pyspark.sql.types import StructType, StructField, StringType, TimestampType, IntegerType

# 創(chuàng)建SparkSession實(shí)例（流處理程序的入口點(diǎn)）
spark = SparkSession.builder \
    .appName("RealTimeRiskEngine") \  # 設(shè)置應(yīng)用名稱
    .config("spark.sql.shuffle.partitions", "6") \  # 優(yōu)化分區(qū)數(shù)
    .getOrCreate()  # 獲取或創(chuàng)建會(huì)話實(shí)例

# 假設(shè)事件數(shù)據(jù)模式（實(shí)際應(yīng)用中根據(jù)業(yè)務(wù)定義）
event_schema = StructType([
    StructField("user_id", StringType(), True),       # 用戶ID
    StructField("event_type", StringType(), True),    # 事件類型（登錄、支付等）
    StructField("ip_address", StringType(), True),    # IP地址
    StructField("device_id", StringType(), True),     # 設(shè)備ID
    StructField("location_city", StringType(), True), # 城市位置
    StructField("event_time", TimestampType(), True)  # 事件時(shí)間戳
])

# ===== 創(chuàng)建事件流源 =====
# 假設(shè)從Kafka讀取事件數(shù)據(jù)（實(shí)際源可能是Kafka、Kinesis等）
events_df = spark.readStream \
    .format("kafka") \  # 數(shù)據(jù)源格式
    .option("kafka.bootstrap.servers", "kafka-risk:9092") \  # Kafka集群
    .option("subscribe", "user_events") \  # 訂閱主題
    .load() \
    .select(
        from_json(col("value").cast("string"), event_schema).alias("data")  # 解析JSON
    ).select("data.*")  # 展平結(jié)構(gòu)

# 添加水印處理延遲數(shù)據(jù)（基于事件時(shí)間）
events_df = events_df.withWatermark("event_time", "10 minutes")

# ===== 特征工程 =====
# 計(jì)算設(shè)備變更次數(shù)（基于用戶會(huì)話）
device_change_df = events_df.groupBy(
    "user_id",
    window("event_time", "1 hour")  # 1小時(shí)滾動(dòng)窗口
).agg(
    count("device_id").alias("device_count"),  # 設(shè)備使用次數(shù)
    expr("count(distinct device_id)").alias("distinct_devices")  # 不同設(shè)備數(shù)
).withColumn(
    "device_change", 
    expr("CASE WHEN distinct_devices > 1 THEN 1 ELSE 0 END")  # 設(shè)備變更標(biāo)志
)

# 計(jì)算城市變更次數(shù)（類似設(shè)備變更）
city_change_df = events_df.groupBy(
    "user_id",
    window("event_time", "1 hour")
).agg(
    expr("count(distinct location_city)").alias("distinct_cities")
).withColumn(
    "city_change", 
    expr("CASE WHEN distinct_cities > 1 THEN 1 ELSE 0 END")
)

# 計(jì)算登錄次數(shù)（1小時(shí)內(nèi)）
login_count_df = events_df.filter("event_type = 'login'") \  # 僅登錄事件
    .groupBy(
        "user_id",
        window("event_time", "1 hour")
    ).agg(count("*").alias("login_count"))

# 合并特征數(shù)據(jù)集
feature_df = device_change_df.join(
    city_change_df, 
    ["user_id", "window"], 
    "left_outer"  # 左外連接確保所有用戶
).join(
    login_count_df, 
    ["user_id", "window"], 
    "left_outer"
).fillna(0)  # 填充空值為0

# ===== 規(guī)則引擎 =====
# 定義風(fēng)險(xiǎn)規(guī)則集（業(yè)務(wù)邏輯）
# 格式：條件 -> 風(fēng)險(xiǎn)等級(jí)
risk_rules = [
    "(device_change = 1 AND city_change = 1) -> 'HIGH_RISK'",  # 規(guī)則1：設(shè)備+城市同時(shí)變更
    "(login_count > 5) -> 'MEDIUM_RISK'",  # 規(guī)則2：1小時(shí)內(nèi)登錄超過5次
    # 默認(rèn)規(guī)則（無匹配時(shí)）
    "1 = 1 -> 'LOW_RISK'"  # 默認(rèn)低風(fēng)險(xiǎn)
]

# 構(gòu)建CASE表達(dá)式
case_expr = "CASE "
for rule in risk_rules:
    # 分割規(guī)則為條件和風(fēng)險(xiǎn)等級(jí)
    condition, risk_level = rule.split("->")
    # 添加到CASE表達(dá)式
    case_expr += f"WHEN {condition.strip()} THEN '{risk_level.strip()}' "
case_expr += "END"

# 應(yīng)用風(fēng)險(xiǎn)規(guī)則引擎
risk_df = feature_df.withColumn(
    "risk_level",  # 新增風(fēng)險(xiǎn)等級(jí)列
    expr(case_expr)  # 執(zhí)行規(guī)則引擎
).select(  # 選擇關(guān)鍵字段
    "user_id", 
    "window.start", 
    "window.end", 
    "device_change", 
    "city_change", 
    "login_count", 
    "risk_level"
)

# ===== 輸出到風(fēng)險(xiǎn)數(shù)據(jù)庫 =====
# 將風(fēng)險(xiǎn)評(píng)估結(jié)果寫入Elasticsearch
risk_query = risk_df.writeStream \
    .outputMode("update") \  # 更新模式（只輸出變更記錄）
    .format("org.elasticsearch.spark.sql") \  # Elasticsearch連接器
    .option("es.nodes", "es1:9200,es2:9200") \  # ES集群節(jié)點(diǎn)
    .option("es.resource", "risk_events") \  # ES索引/類型（ES7+使用索引名）
    .option("es.mapping.id", "user_id") \  # 文檔ID字段（基于用戶ID）
    .option("es.write.operation", "upsert") \  # 更新插入模式
    .option("checkpointLocation", "/checkpoints/risk_engine") \  # 檢查點(diǎn)目錄（容錯(cuò)）
    .trigger(processingTime="1 minute") \  # 每分鐘觸發(fā)
    .start()  # 啟動(dòng)流處理

# 同時(shí)輸出到控制臺(tái)用于調(diào)試
console_query = risk_df.writeStream \
    .outputMode("update") \
    .format("console") \
    .option("truncate", "false") \
    .start()

# 等待流處理終止
spark.streams.awaitAnyTermination()

# 生產(chǎn)環(huán)境應(yīng)添加優(yōu)雅停止邏輯

# 導(dǎo)入必要的PySpark模塊
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, count, col
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

# 初始化Spark會(huì)話（配置反壓和檢查點(diǎn)）
spark = SparkSession.builder \
    .appName("IPRegistrationFraudDetection") \
    .config("spark.sql.shuffle.partitions", "8") \  # 根據(jù)集群規(guī)模調(diào)整
    .config("spark.streaming.backpressure.enabled", "true") \  # 啟用反壓
    .config("spark.streaming.kafka.maxRatePerPartition", "1000") \  # 每分區(qū)最大速率
    .getOrCreate()

# 定義注冊(cè)事件的數(shù)據(jù)結(jié)構(gòu)
registration_schema = StructType([
    StructField("user_id", StringType(), True),      # 注冊(cè)用戶ID
    StructField("ip", StringType(), True),          # 注冊(cè)IP地址
    StructField("device_id", StringType(), True),   # 設(shè)備標(biāo)識(shí)
    StructField("event_time", TimestampType(), True) # 事件時(shí)間（必須時(shí)間戳類型）
])

# ===== 數(shù)據(jù)源配置 =====
# 從Kafka讀取注冊(cè)事件流（生產(chǎn)環(huán)境配置）
registrations = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092") \
    .option("subscribe", "user_registrations") \  # 訂閱注冊(cè)主題
    .option("startingOffsets", "latest") \  # 從最新位置開始
    .option("failOnDataLoss", "false") \  # 容忍數(shù)據(jù)丟失
    .load() \
    .select(
        from_json(col("value").cast("string"), registration_schema).alias("data")
    ).select("data.*")  # 提取結(jié)構(gòu)化數(shù)據(jù)

# 添加水印處理延遲數(shù)據(jù)（10分鐘延遲）
registrations = registrations.withWatermark("event_time", "10 minutes")

# ===== 核心檢測(cè)邏輯 =====
# 每10分鐘窗口統(tǒng)計(jì)每個(gè)IP的注冊(cè)次數(shù)
ip_registration_counts = registrations.groupBy(
    window("event_time", "10 minutes"),  # 10分鐘滾動(dòng)窗口
    "ip"                                # 按IP分組
).agg(
    count("*").alias("registration_count")  # 計(jì)算注冊(cè)次數(shù)
)

# 過濾出異常IP（10分鐘內(nèi)注冊(cè)超過20次）
suspicious_ips = ip_registration_counts.filter(
    col("registration_count") > 20  # 閾值可根據(jù)業(yè)務(wù)調(diào)整
).select(
    col("window.start").alias("window_start"),  # 窗口開始時(shí)間
    col("window.end").alias("window_end"),      # 窗口結(jié)束時(shí)間
    col("ip"),                                  # 嫌疑IP
    col("registration_count")                   # 注冊(cè)次數(shù)
)

# ===== 輸出配置 =====
# 方案1：輸出到控制臺(tái)（調(diào)試用）
console_query = suspicious_ips.writeStream \
    .outputMode("complete") \  # 完整模式（顯示所有結(jié)果）
    .format("console") \
    .option("truncate", "false") \
    .option("numRows", 100) \
    .trigger(processingTime="1 minute") \  # 每分鐘觸發(fā)一次
    .start()

# 方案2：輸出到Elasticsearch（生產(chǎn)環(huán)境）
es_query = suspicious_ips.writeStream \
    .outputMode("complete") \
    .format("org.elasticsearch.spark.sql") \
    .option("es.nodes", "es1:9200") \
    .option("es.resource", "fraud_ips") \  # 索引名稱
    .option("es.mapping.id", "ip") \       # 使用IP作為文檔ID
    .option("es.write.operation", "upsert") \
    .option("checkpointLocation", "/checkpoints/ip_fraud") \  # 檢查點(diǎn)目錄
    .start()

# 方案3：輸出到Kafka告警主題（生產(chǎn)環(huán)境）
alert_query = suspicious_ips.selectExpr(
    "CAST(ip AS STRING) AS key",
    "to_json(struct(*)) AS value"
).writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka1:9092") \
    .option("topic", "fraud_alerts") \
    .option("checkpointLocation", "/checkpoints/ip_fraud_kafka") \
    .start()

# 等待任意流查詢終止
spark.streams.awaitAnyTermination()

# 生產(chǎn)環(huán)境應(yīng)添加信號(hào)捕獲和優(yōu)雅關(guān)閉邏輯

# server.properties
num.network.threads=16  # 網(wǎng)絡(luò)線程池
num.io.threads=32       # 磁盤IO線程
log.flush.interval.messages=10000 
socket.send.buffer.bytes=1024000 # 發(fā)送緩沖區(qū)

# 集群資源配置示例
spark-submit --master yarn \
    --num-executors 16 \        # 執(zhí)行器數(shù)量
    --executor-cores 4 \        # 每執(zhí)行器內(nèi)核
    --executor-memory 8g \      # 執(zhí)行器內(nèi)存
    --conf spark.sql.shuffle.partitions=128 \  # 并行度
    --conf spark.streaming.backpressure.enabled=true  # 反壓