# 原始SQL（存在数据倾斜）
hive> SELECT product_id, count(*) as click_count
FROM fgedu.user_behavior
GROUP BY product_id;

# 执行结果（输出日志）
…
Stage-1 map = 100%, reduce = 0%, Cumulative CPU 10.2 sec
Stage-1 map = 100%, reduce = 10%, Cumulative CPU 15.6 sec
Stage-1 map = 100%, reduce = 20%, Cumulative CPU 20.3 sec
# 卡住，部分Reduce任务处理热点数据

# 优化后SQL
hive> SELECT product_id, sum(click_count) as total_clicks
FROM (
SELECT
CASE WHEN product_id = ‘hot_product_123’ THEN
concat(product_id, ‘_’, rand())
ELSE product_id END AS new_product_id,
count(*) as click_count
FROM fgedu.user_behavior
GROUP BY new_product_id
) t
GROUP BY product_id;

# 执行结果（输出日志）
…
Stage-1 map = 100%, reduce = 100%, Cumulative CPU 35.2 sec
Stage-2 map = 100%, reduce = 100%, Cumulative CPU 12.5 sec
MapReduce Total cumulative CPU time: 47 seconds 700 msec
OK
product_id total_clicks
hot_product_123 1258000
product_456 23500
product_789 18900
…
更多视频教程www.fgedu.net.cn

# 原始代码（存在数据倾斜）
[root@fgedu.net.cn ~]# spark-submit –master yarn –deploy-mode cluster \
–class com.fgedu.spark.TransactionAnalysis \
transaction-analysis.jar

# 执行结果（输出日志）
…
Stage 1:=====================================> (99 + 1) / 100
# 最后一个task执行时间特别长

# 优化后代码
[root@fgedu.net.cn ~]# vi TransactionAnalysis.scala
val spark = SparkSession.builder()
.appName(“TransactionAnalysis”)
.getOrCreate()

import spark.implicits._

val transactions = spark.read.parquet(“hdfs://fgedu.net.cn:9000/user/fgedu/transactions”)

// 处理数据倾斜
val processedData = transactions.rdd.map(row => {
val userId = row.getString(0)
val amount = row.getDouble(1)
// 对热点用户添加随机前缀
val hotUsers = Set(“user_1001”, “user_2002”, “user_3003”)
if (hotUsers.contains(userId)) {
(userId + “_” + scala.util.Random.nextInt(10), amount)
} else {
(userId, amount)
}
})

val result = processedData.reduceByKey(_ + _).map { case (key, value) =>
// 移除随机前缀
val userId = if (key.contains(“_”)) key.split(“_”)(0) else key
(userId, value)
}.reduceByKey(_ + _)

result.saveAsTextFile(“hdfs://fgedu.net.cn:9000/user/fgedu/result”)

# 执行结果（输出日志）
…
Stage 1:========================================> (100 + 0) / 100
Stage 2:========================================> (100 + 0) / 100
Job completed successfully
Total time taken: 2 minutes 35 seconds
学习交流加群风哥微信: itpux-com

# 1. 使用Map端Join
[root@fgedu.net.cn ~]# hadoop jar /bigdata/app/hadoop/share/hadoop/tools/lib/hadoop-streaming-3.3.4.jar \
-D mapreduce.job.reduces=10 \
-D mapreduce.map.output.compress=true \
-D mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.SnappyCodec \
-file mapper.py -mapper “python mapper.py” \
-file reducer.py -reducer “python reducer.py” \
-input /user/fgedu/orders /user/fgedu/customers \
-output /user/fgedu/join_result

# 2. 实现Map端Join的mapper.py
[root@fgedu.net.cn ~]# vi mapper.py
#!/usr/bin/env python
# mapper.py
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: `http://www.fgedu.net.cn`

import sys
import json

# 加载小表到内存
customers = {}
with open(‘customers.txt’, ‘r’) as f:
for line in f:
parts = line.strip().split(‘\t’)
if len(parts) == 2:
customers[parts[0]] = parts[1]

for line in sys.stdin:
line = line.strip()
if not line:
continue
parts = line.split(‘\t’)
if len(parts) >= 2:
if parts[0].startswith(‘order’):
# 订单数据
order_id, customer_id, amount = parts[0], parts[1], parts[2]
if customer_id in customers:
print(‘\t’.join([order_id, customer_id, customers[customer_id], amount]))

# 执行结果（输出日志）
…
MapReduce Total cumulative CPU time: 1 minute 45 seconds
OK
order_001 customer_1001 Alice 100.50
order_002 customer_1001 Alice 200.75
order_003 customer_2002 Bob 150.25
…
更多学习教程公众号风哥教程itpux_com