Разделите фрейм данных Spark на несколько фреймов, а затем снова присоединитесь - PullRequest
Новая
       46

Разделите фрейм данных Spark на несколько фреймов, а затем снова присоединитесь

0 голосов
/ 27 сентября 2019

У меня есть искровой фрейм данных, который выбирает данные из таблицы кустов cloudera, с примерно 4 тысячами переменных, я делю их на четыре различных фрейма данных (определенные столбцы) и выполняю некоторые операции (замените Null некоторым числом в каждом фрейме данных отдельно)и затем объедините все кадры данных (часть соединения).

final2 - это искровой кадр данных с четырьмя тысячами столбцов и 1,5 миллионами записей.Код работает, но это занимает огромное время, Также я запускаю код по частям, и замена нуля не занимает времени, Но когда я объединяю все кадры данных, это занимает огромное время. Пожалуйста, помогите, если вокруг нет.Спасибо. 

  final2 = spark.sql("SELECT * FROM prod_mytable ) 
  final2 =final.dtypes  
  list_final2        = [t[0] for t in final2]  ##3977

  list_int_f        = [t[0] for t in final2 if t[1] == 'int']  
  list_double_f     = [t[0] for t in final2 if t[1] == 'double']  
  list_bigint_f     = [t[0] for t in final2 if t[1] == 'bigint']  
  list_num_f        = list_int_f + list_double_f + list_bigint_f


  list_inactive     = [col for col in final.columns if 
  ('max_consecutive_inactive_days_' in col )]
  list_inactive_3m  = [col for col in final.columns if 
  ('max_consecutive_inactive_days_' in col and '_3m' in col)] ##4
  list_inactive_m   = [col for col in final.columns if 
  ('max_consecutive_inactive_days_' in col and '_m' in col)]  ##12
  list_inactive_w   = [col for col in final.columns if 
  ('max_consecutive_inactive_days_' in col and '_w' in col)]  ##48


  list_num_f2        = [item for item in list_num_f if item not in 
  list_inactive] 

  list_final22      = [item for item in list_final2 if item not in 
  list_num_f]  



  list_num_f2             = l_vf + list_num_f2
  list_inactive_3m        = l_vf + list_inactive_3m
  list_inactive_m         = l_vf + list_inactive_m
  list_inactive_w         = l_vf + list_inactive_w

  list_final22_df            = final.select(list_final22)
  list_num_f2_df             = final.select(list_num_f2 )
  list_inactive_3m_df        = final.select(list_inactive_3m)
  list_inactive_m_df         = final.select(list_inactive_m )
  list_inactive_w_df         = final.select(list_inactive_w )



  list_num_f2_df        = list_num_f2_df.na.fill(0) 
  list_inactive_3m_df   =list_inactive_3m_df.na.fill(90) 
  list_inactive_m_df    =list_inactive_m_df.na.fill(30) 
  list_inactive_w_df    =list_inactive_w_df.na.fill(7) 



  final33 = (list_final22_df.join(list_num_f2_df,
                                         "mobil_number",
                                        how = "left").join(
                                  list_inactive_3m_df,
                                         "mobil_number",
                                        how = "left").join(
                                  list_inactive_m_df,
                                         "mobil_number",
                                        how = "left").join(
                                  list_inactive_w_df,
                                         "mobil_number",
                                        how = "left"))

  df_314 = final33.filter(final33.mobil_number == '123456789') 
  df_314.show()
...