Я пытаюсь count частые наборы элементов, генерируемые ростом FP mllib с использованием Spark API.My Spark - версия 1.5.1.Ниже приведен мой код:
#!/usr/bin/python
from pyspark.mllib.fpm import FPGrowth
from pyspark import SparkContext,SparkConf
from pyspark import HiveContext
import os
os.environ['PYSPARK_PYTHON']='/usr/bin/python'
appName = "FP_growth"
sc = SparkContext()
sql_context = HiveContext(sc)
def read_spu(prod):#prod_code):
sql = """
select
t.orderno_nosplit,
t.prod_code,
t.item_code,
sum(t.item_qty) as item_qty
from ioc_fdm.fdm_dwr_ioc_fcs_pk_spu_item_f_chain t
where t.prod_code='%s'
group by t.prod_code, t.orderno_nosplit, t.item_code """%prod
spu_result = sql_context.sql(sql)
return spu_result.cache()
if __name__ == '__main__':
spu=read_spu('6727780')
conf=0.7
trans=spu.rdd.repartition(100).map(lambda x: (x[0],x[2])).groupByKey().mapValues(list).values().cache()
model = FPGrowth.train(trans, 0.01, 100)
freq_count = model.freqItemsets().count()
print 'freq_count:',freq_count
sc.stop()
Входные данные считываются из Hadoop, и данные не очень большие, всего около 20000 строк.Однако скрипт работает очень-очень медленно на этапе .count.Я не знаю почему.Из производительности, кажется, это из-за перекоса данных.Но выходные данные не велики (всего около 100 КБ на задачу).
Кластер имеет 8 узлов по 320 ядер и 1,56 Т общей памяти (не только один пользователь).Мой скрипт spark-submit: spark-submit --master yarn-cluster --executor-memory 30g --num-executors 20 --executor-cores 5 FP_growth.py
Вложения - это отпечатки экрана производительности при запуске:
Используемый ресурс
Активные стадии
Задачи