Spark 1.5.1中FP-Growth频繁项集count操作性能缓慢问题排查

Hey there! Let's break down why your freqItemsets().count() is crawling even with just 20k rows of data. Based on your code, cluster setup, and the symptoms you described (suspected skew but small task outputs), here are targeted fixes and checks:

1. Fix Your Data Preprocessing First #

Your pipeline groups order items using groupByKey after pulling data from Hive. In Spark 1.5.1, groupByKey is notoriously inefficient for aggregations, and it’s a common source of hidden skew—even if your final task outputs are tiny, the shuffle phase might be choking on lopsided intermediate data.

Quick fixes for preprocessing: #

• Swap groupByKey for aggregateByKey (it reduces shuffle data by combining values locally first):

  # Replace your existing groupByKey logic with this
  trans = spu.rdd.repartition(100).map(lambda x: (x[0], x[2])) \
      .aggregateByKey([], lambda acc, val: acc + [val], lambda acc1, acc2: acc1 + acc2) \
      .values().cache()
  

• Check for "super orders": If one order has way more items than others, it’ll skew the shuffle. Run this quick check to spot outliers:

  order_item_counts = spu.rdd.map(lambda x: (x[0], 1)).reduceByKey(lambda a, b: a + b)
  # Print top 10 orders with the most items
  print("Top 10 orders by item count:", order_item_counts.takeOrdered(10, key=lambda x: -x[1]))
  

  If you find an order with hundreds/thousands of items, consider splitting it or filtering it out temporarily to test if skew is the issue.

2. Tune FP-Growth Parameters for Your Data Size #

Spark 1.5.1’s FP-Growth implementation is pretty basic compared to newer versions, so parameter choices matter a lot:

• Too many partitions: You set numPartitions=100 in FPGrowth.train, but with only 20k rows, this creates way too many tiny tasks. Task scheduling overhead can eat up more time than the actual computation. Try dropping this to 20 (matching your executor count) or 40.
• Too low minSupport: A minSupport of 0.01 means any itemset that appears 200+ times gets kept. This could generate massive numbers of frequent itemsets, making the count() stage process way more data than you expect. Test with a higher value (like 0.05) first to see if the count speeds up, then adjust to meet your business needs.

3. Optimize Cluster Resource Usage #

Your spark-submit config (--num-executors 20 --executor-cores 5 --executor-memory 30g) looks reasonable, but let’s make sure you’re using resources effectively in a shared cluster:

• Check cluster idle resources: If other users are hogging cores/memory, your tasks might be stuck waiting. Pop into the Yarn ResourceManager UI to see how much capacity is available.
• Verify cache effectiveness: You’re caching spu_result and trans, but make sure the cache is actually in memory. Check the Spark UI’s Storage tab—if data is spilling to disk, that’ll slow down every step that uses those RDDs.
• Tweak shuffle settings: Spark 1.5.1’s default shuffle config can be fragile. Add these to your spark-submit command to reduce retries and memory pressure:

  --conf spark.shuffle.reduceLocality.enabled=false \
  --conf spark.shuffle.io.maxRetries=10 \
  --conf spark.reducer.maxSizeInFlight=96m
  

4. Diagnose the Count Stage Directly #

Even without seeing your screenshots, you can use the Spark UI’s Stages tab to get to the bottom of the slow count:

• Look for tasks that take way longer than others (this is the classic skew red flag). If one task runs for minutes while others finish in seconds, you’ve got skew in the freqItemsets RDD.
• If skew is the issue, try salting the keys to break up the hot partition before counting:

  # Add a salt to split hot keys across partitions
  salted_freq_itemsets = model.freqItemsets().map(lambda x: (hash(x.items) % 20, x))
  # Count per salt, then sum the totals
  total_count = salted_freq_itemsets.groupByKey().map(lambda x: len(x[1])).reduce(lambda a, b: a + b)
  print('freq_count:', total_count)
  

内容的提问来源于stack exchange，提问作者Wilber