pysparkでデータフレームの辞書を作成する

Question

年と月の辞書を作成しようとしています。そのマクロの一種は、私が必要とするものではありません。年月のそれから私は、私はパンダでこれを行うことができ、以下に示すように、単一の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 

dfをpysparkに動的な列を追加しているとき、私は挑戦に直面していますが、私はpysparkで行う必要があり

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

動的なデータフレームの辞書を生成し、列を動的に追加して条件を実行し、年単位のデータフレームを生成して単一のデータフレームにマージする考え方はありますか？どんな助けもありがとう。私はコードの下にしようとしている

Answer 1

はみとめデータフレームを追加するには罰金

// 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 

を働いているコードの下に使用します。それは働いた

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