кэширование фреймов данных pyspark не приводит к увеличению производительности - PullRequest
Новая
       20

кэширование фреймов данных pyspark не приводит к увеличению производительности

0 голосов
/ 16 апреля 2020

Я пытаюсь сделать свои сценарии более эффективными.

На данный момент у меня есть 10 сценариев - все они считывают данные, обрабатывают и выводят их.

Все они читают из хотя те же основные таблицы БД и просто делают разные вещи с данными.

Так что я объединил в один скрипт, с идеей, что я читаю данные только один раз, а не 10 раз.

Разве это не должно привести к более быстрому выполнению? Потому что это не так.

Ниже приведен пример структуры, которую я использую

Любая помощь будет удивительной

Спасибо 

'''
TABLE DEFINITION AND CACHING
'''
spark_session = create_session('usage CF')
usage_logs = spark_session.sql("Select * from db.table where dt = " + yday_date ).cache()
user_logs = spark_session.sql("Select * from db2.table2 where dt = " + yday_date ).cache()
usercat = spark_session.sql("Select * from db3.table3 where dt = " + yday_date ).cache()
radius_logs = spark_session.sql("Select * from db.table4 where dt = " + yday_date )
radius = radius_logs.select('emsisdn2', 'sessionid2', 'custavp1').cache()


'''
usage CF
'''
usage = usage_logs.select('field1', 'field2', 'field3')
conditions = [usage.sid == radius.sessionid2]
df3 = usage.join(radius, conditions, how='left')
df4 = df3.groupBy('field1', 'field2').agg(sqlfunc.sum('field3').alias('bytesdl'))
usage = df4.createOrReplaceTempView('usage')
usage_table_output = spark_session.sql(' insert overwrite table outputdb.outputtbl partition(dt = ' + yday_date + ') select "usage" as type, * from usage ')

'''
user CF
'''
user = usage_logs.filter((usage_logs.vslsessid == '0')).select('field1', 'field2', 'field3', 'field4')
conditionsx = [user.sessionid == radius.sessionid2]
user_joined = user.join(radius, conditionsx, how='left')
user_output = user_joined.groupBy('field1', 'field2', 'field3').agg(sqlfunc.sum('field4').alias('bytesdl'))
user = user_output.createOrReplaceTempView('user')
user_table_output = spark_session.sql(' insert overwrite table outputdb.outputtbl2 partition(dt = ' + yday_date + ') select "user" as type, * from user')
...