foreachPartition内でSQLContextとSparkContextを使用する方法

Question

foreachPartitionの中でSparkContextとSQLContextを使用したいが、シリアル化エラーのためにできない。 ..私は、両方のオブジェクトが直列化可能でないことを知っているが、私はforeachPartitionがスパークコンテキストとSQLContextの両方が利用可能なマスター、上で実行されていることを考え

表記：これは私の現在のコードが

`msg -> Map[String,String]` 
`result -> Iterable[Seq[Row]]` 

ある （UtilsDMは、extends Serializableというオブジェクトです）。失敗したコード部分はval schema =...から始まり、をDataFrameに書き込んで、それをParquetに保存します。私がコードを編成した方法が非効率なのかもしれませんが、私はここにあなたの推薦をしたいと思います。ありがとう。

// Here I am creating df from parquet file on S3 
val exists = FileSystem.get(new URI("s3n://" + bucketNameCode), sc.hadoopConfiguration).exists(new Path("s3n://" + bucketNameCode + "/" + pathToSentMessages)) 
var df: DataFrame = null 
if (exists) { 
    df = sqlContext 
    .read.parquet("s3n://bucket/pathToParquetFile") 
} 
UtilsDM.setDF(df) 

// Here I process myDStream 
myDStream.foreachRDD(rdd => { 
    rdd.foreachPartition{iter => 
    val r = new RedisClient(UtilsDM.getHost, UtilsDM.getPort) 
    val producer = UtilsDM.createProducer 
    var df = UtilsDM.getDF 
    val result = iter.map{ msg => 
     // ... 
     Seq(msg("key"),msg("value")) 
    } 

    // HERE I WANT TO WRITE result TO S3, BUT IT FAILS 
    val schema = StructType(
        StructField("key", StringType, true) :: 
        StructField("value", StringType, true) 

    result.foreach { row => 
     val rdd = sc.makeRDD(row) 
     val df2 = sqlContext.createDataFrame(rdd, schema) 

     // If the parquet file is not created, then create it 
     var df_final: DataFrame = null 
     if (df != null) { 
      df_final = df.unionAll(df2) 
     } else { 
      df_final = df2 
     } 
     df_final.write.parquet("s3n://bucket/pathToSentMessages) 
} 
    } 
}) 

EDIT：

私はスパーク1.6.2とScalaの2.10.6を使用しています。

Answer 1

これはできません。 SparkContext,SQLContextおよびSparkSessionは、ドライバでのみ使用できます。あなたはforeachRDDのトップレベルでsqlContextを使用することができます。

myDStream.foreachRDD(rdd => { 
    val df = sqlContext.createDataFrame(rdd, schema) 
    ... 
}) 

あなたは、変換/アクションでそれを使用することはできません。

myDStream.foreachRDD(rdd => { 
    rdd.foreach { 
     val df = sqlContext.createDataFrame(...) 
     ... 
    } 
}) 

あなたはおそらく相当します：

myDStream.foreachRDD(rdd => { 
    val foo = rdd.mapPartitions(iter => doSomethingWithRedisClient(iter)) 
    val df = sqlContext.createDataFrame(foo, schema) 
    df.write.parquet("s3n://bucket/pathToSentMessages) 
}) 

Answer 2

私は、ループ内の既存のSparkContextを（私は事前にsparkContext SCを作成していると仮定）を使用すると、その火花のバージョンを使用している、すなわち

// this works 
stream.foreachRDD(_ => { 
    // update rdd 
    .... = SparkContext.getOrCreate().parallelize(...) 
}) 

// this doesn't work - throws a SparkContext not serializable error 
stream.foreachRDD(_ => { 
    // update rdd 
    .... = sc.parallelize(...) 
})