RDDのキーで指定されたパスにrddの値を書き込む

Question

keyに指定されたatのパスに値に格納された値のパーケットファイルを作成したいとします。パス/ユーザー/ dir_aでそう

#key is path where the value is to written 
rdd = sc.parallelize([('/user/dir_a','tableA'),('/user/dir_b','tableB'),('/user/dir_c','tableC')]) 

、Aは私が何をしたか

を書かれている：

def writeToHdfs(x): 
    path = x[0] 
    outputpath = OUT_DIR + path 
    log.info('Creating dataframe') 
    s = SparkSession(sc) 
    df = s.createDataFrame(x[1], schema)) 
    df.write.parquet(outputpath) 

rdd.foreach(writeToHdfs) 

感謝。

Answer 1

私は、このシナリオのためのボックスソリューションはないと信じています。コードはScalaにありますが、ロジックはPythonでも同じです。

val baseRDD = sc.parallelize(Seq(("/user/dir_a", "tableA"), ("/user/dir_b", "tableB"), ("/user/dir_c", "tableC"))).cache() 

    val groupedRDD = baseRDD.groupByKey() 

    //Bring the keys to driver.Its little expensive operation 
    // but we need keys(paths) after all. 
    val keys = groupedRDD.keys.collect() 

    //Create RDDs specific to ur paths 
    val rddList = keys.map { key => 

     val rdd = baseRDD.filter(f => f._1.==(key)) 

     (key, rdd) 
    } 

    //Now you have list of RDDs specific to paths. iterate each RDD and save them to file 
    rddList.foreach(f => { 

     val path = f._1 
     f._2.values.saveAsTextFile(path) 
    }) 

注：パフォーマンスを向上させる必要があると思われる場合は、RDDをキャッシュしてください。 saveAsTextFile(...)をそれぞれの方法に置き換えてください。