创建数据帧的字典中pyspark

我试图创建一个字典年份和月份。它是一种我可以通过不需要的宏调。年和月。我现在面临的挑战，而在pyspark添加动态列DF创建数据帧的字典中pyspark

df = spark.createDataFrame([(1, "foo1",'2016-1-31'),(1, "test",'2016-1-31'), (2, "bar1",'2012-1-3'),(4, "foo2",'2011-1-11')], ("k", "v","date")) 
w = Window().partitionBy().orderBy(col('date').desc())   
df = df.withColumn("next_date",lag('date').over(w).cast(DateType())) 
df = df.withColumn("next_name",lag('v').over(w)) 
df = df.withColumn("next_date",when(col("k") != lag(df.k).over(w),date_add(df.date,605)).otherwise(col('next_date'))) 
df = df.withColumn("next_name",when(col("k") != lag(df.k).over(w),"").otherwise(col('next_name'))) 

import copy 
dict_of_YearMonth = {} 

for yearmonth in [200901,200902,201605 .. etc]: 

    key_name = 'Snapshot_'+str(yearmonth) 
    dict_of_YearMonth[key_name].withColumn("test",yearmonth) 
    dict_of_YearMonth[key_name].withColumn("test_date",to_date(''+yearmonth[:4]+'-'+yearmonth[4:2]+'-1'+'')) 
# now i want to add a condition 
    if(dict_of_YearMonth[key_name].test_date >= dict_of_YearMonth[key_name].date) and (test_date <= next_date) then output snapshot_yearmonth /// i.e dataframe which satisfy this condition i am able to do it in pandas but facing challenge in pyspark 
dict_of_YearMonth[key_name] 
dict_of_YearMonth

然后我想连接所有的数据帧为单pyspark数据帧，我可以在大熊猫做到这一点，如下图所示，但我需要在pyspark做

snapshots=pd.concat([dict_of_YearMonth['Snapshot_201104'],dict_of_YearMonth['Snapshot_201105']])

如果是任何其他的想法产生与动态地添加列的动态数据帧的字典和执行情况，并基于年度数据帧，并在单独的数据帧合并。任何帮助，将不胜感激。

来源

2017-03-09 user_tom

我曾尝试下面的代码工作正常

// Function to append all the dataframe using union 
def unionAll(*dfs): 
return reduce(DataFrame.unionAll, dfs) 

// convert dates 
def is_date(x): 
    try: 
     x= str(x)+str('01') 
     parse(x) 
     return datetime.datetime.strptime(x, '%Y%m%d').strftime("%Y-%m-%d") 
    except ValueError: 
     pass # if incorrect format, keep trying other format 

dict_of_YearMonth = {} 
for yearmonth in [200901,200910]: 
key_name = 'Snapshot_'+str(yearmonth) 
dict_of_YearMonth[key_name]=df 
func = udf(lambda x: yearmonth, StringType()) 
dict_of_YearMonth[key_name] = df.withColumn("test",func(col('v'))) 
default_date = udf (lambda x : is_date(x)) 
dict_of_YearMonth[key_name] = dict_of_YearMonth[key_name].withColumn("test_date",default_date(col('test')).cast(DateType())) 
dict_of_YearMonth

要添加多发dataframes使用下面的代码：

final_df = unionAll(dict_of_YearMonth['Snapshot_200901'], dict_of_YearMonth['Snapshot_200910'])

来源

2017-03-09 19:34:58 user07

感谢它的工作！ –

创建数据帧的字典中pyspark

回答

相关问题