# 1. 数据准备
scala> spark.sql(“””
| CREATE TABLE IF NOT EXISTS fgedu_ads.ads_user_behavior_analysis (
| user_id BIGINT,
| total_pv BIGINT,
| total_buy BIGINT,
| total_cart BIGINT,
| total_fav BIGINT,
| active_days INT,
| avg_daily_actions DOUBLE,
| user_level STRING,
| dt STRING
| )
| USING PARQUET
| PARTITIONED BY (dt STRING)
| “””)
res0: org.apache.spark.sql.DataFrame = []

# 2. 用户行为分析
scala> spark.sql(“””
| INSERT INTO TABLE fgedu_ads.ads_user_behavior_analysis PARTITION(dt=’2026-04-08′)
| SELECT
| user_id,
| SUM(pv_count) as total_pv,
| SUM(buy_count) as total_buy,
| SUM(cart_count) as total_cart,
| SUM(fav_count) as total_fav,
| COUNT(*) as active_days,
| AVG(pv_count + buy_count + cart_count + fav_count) as avg_daily_actions,
| CASE
| WHEN SUM(buy_count) >= 10 THEN ‘VIP’
| WHEN SUM(buy_count) >= 5 THEN ‘Active’
| WHEN SUM(pv_count) >= 100 THEN ‘Potential’
| ELSE ‘Normal’
| END as user_level
| FROM fgedu_dws.dws_user_behavior_summary
| WHERE dt BETWEEN ‘2026-04-01’ AND ‘2026-04-08′
| GROUP BY user_id
| “””)
res1: org.apache.spark.sql.DataFrame = []

# 3. 查看分析结果
scala> spark.sql(“””
| SELECT user_level, COUNT(*) as user_count, AVG(total_buy) as avg_buy
| FROM fgedu_ads.ads_user_behavior_analysis
| WHERE dt=’2026-04-08’
| GROUP BY user_level
| ORDER BY user_count DESC
| “””).show()
+———-+———-+——————+
|user_level|user_count| avg_buy|
+———-+———-+——————+
| Normal| 500000| 0.0|
| Potential| 300000| 0.5|
| Active| 150000| 6.5|
| VIP| 50000| 15.2|
+———-+———-+——————+

# 1. 实时PV/UV统计
scala> val pvuvQuery = spark.readStream
| .format(“kafka”)
| .option(“kafka.bootstrap.servers”, “192.168.1.62:9092”)
| .option(“subscribe”, “fgedu-user-behavior”)
| .load()
| .select(from_json(col(“value”).cast(“string”), schema).as(“data”))
| .select(
| col(“data.user_id”).alias(“user_id”),
| col(“data.behavior_type”).alias(“behavior_type”),
| current_timestamp().alias(“process_time”)
| )
| .withWatermark(“process_time”, “10 seconds”)
| .groupBy(window(col(“process_time”), “1 minute”))
| .agg(
| count(“*”).alias(“pv”),
| countDistinct(“user_id”).alias(“uv”)
| )
| .writeStream
| .outputMode(“update”)
| .format(“console”)
| .start()
pvuvQuery: org.apache.spark.sql.streaming.StreamingQuery = StreamingQuery – pvuv

# 2. 输出示例
——————————————-
Batch: 10
——————————————-
+——————–+——+—–+
| window| pv| uv|
+——————–+——+—–+
|[2026-04-08 13:00…|100000| 5000|
|[2026-04-08 13:01…|120000| 5500|
|[2026-04-08 13:02…| 95000| 4800|
+——————–+——+—–+

# 3. 写入Redis
scala> val redisQuery = pvuvDF.writeStream
| .foreach(new RedisSink())
| .start()

# Redis存储格式
# Key: pvuv:2026-04-08:13:00
# Value: {“pv”:100000,”uv”:5000}

# 排查步骤
# 1. 查看Stage详情
# Spark UI -> Stages -> 查看Task执行时间分布

# 2. 查看数据分布
scala> spark.sql(“””
| SELECT user_id, COUNT(*) as cnt
| FROM fgedu_dwd.dwd_user_behavior_detail
| WHERE dt=’2026-04-08′
| GROUP BY user_id
| ORDER BY cnt DESC
| LIMIT 10
| “””).show()

# 解决方案
# 1. 增加分区数
spark.sql.shuffle.partitions=400

# 2. 使用盐值
scala> val saltedDF = df
| .withColumn(“salt”, (rand() * 10).cast(“int”))
| .withColumn(“salted_key”, concat(col(“user_id”), lit(“_”), col(“salt”)))

# 3. 广播Join
scala> val result = largeDF.join(broadcast(smallDF), “key”)

# 4. 采样倾斜Key单独处理
scala> val skewedKeys = df.groupBy(“key”).count().filter($”count” > 1000000).select(“key”)
scala> val skewedData = df.join(skewedKeys, “key”)
scala> val normalData = df.join(skewedKeys, “key”, “left_anti”)

# 排查步骤
# 1. 查看Executor日志
$ yarn logs -applicationId application_xxx | grep -i “OutOfMemory”

# 2. 查看内存使用
# Spark UI -> Executors -> Memory Used

# 解决方案
# 1. 增加Executor内存
–executor-memory 16g

# 2. 调整内存比例
spark.memory.fraction=0.8
spark.memory.storageFraction=0.5

# 3. 减少分区数据量
spark.sql.shuffle.partitions=400

# 4. 使用广播变量
scala> val broadcastVar = spark.sparkContext.broadcast(largeData)

# 5. 清理缓存
scala> spark.catalog.clearCache()