如何在 foreachPartition 中使用 SQLContext 和 SparkContext答案

【问题标题】：How to use SQLContext and SparkContext inside foreachPartition如何在 foreachPartition 中使用 SQLContext 和 SparkContext
【发布时间】：2018-08-12 02:03:42
【问题描述】：

我想在foreachPartition 中使用 SparkContext 和 SQLContext，但由于序列化错误无法执行。我知道这两个对象都不可序列化，但我认为foreachPartition 是在master 上执行的，其中Spark Context 和SQLContext 都可用..

符号：

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

这是我当前的代码（UtilsDM 是 extends Serializable 的对象）。失败的部分代码从val schema =...开始，我想将result写入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)
}
  }
})

编辑：

我正在使用 Spark 1.6.2 和 Scala 2.10.6。

【问题讨论】：

您使用的是哪个版本的 spark ？
@MRSrinivas：我使用的是 Spark 1.6.2 和 Scala 2.10.6。抱歉没有提及。

标签： scala apache-spark

【解决方案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)
})

【讨论】：

好的，谢谢。这意味着我应该使用mapPartitions而不是foreachPartition来返回result？你能告诉我怎么做吗？
应该是这样吗？它不适合我：myDStream.foreachRDD(rdd => { val finalResult = rdd.mapPartitions(iter => val r = new RedisClient(UtilsDM.getHost, UtilsDM.getPort) val result = iter.map{ msg => ... }) ) })
我不完全理解你的代码，但无论你做什么，你都不能在RDD.foreach、RDD.map、RDD.mapPartitions等中使用sqlContext。
我的意思是这部分没有编译：(iter => doSomethingWithRedisClient(iter)) Coule 你请给我看具体的例子吗？谢谢。
现在编译。还有一个问题。可以在集群中使用像我的情况 UtilsDM 这样的可序列化对象吗？还是这个解决方案只能在本地工作？

【解决方案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(...)
})

【讨论】：

你了解 op 想要实现的目标吗？