Spark 正在重复工作

Question

我正面临 Spark 的一种奇怪行为。这是我的代码：

object MyJob {
    def main(args: Array[String]): Unit = {
        val sc = new SparkContext()
        val sqlContext = new hive.HiveContext(sc)

        val query = "<Some Hive Query>"
        val rawData = sqlContext.sql(query).cache()

        val aggregatedData = rawData.groupBy("group_key")
           .agg(
               max("col1").as("max"),
               min("col2").as("min")
           )

        val redisConfig =  new RedisConfig(new RedisEndpoint(sc.getConf))
        aggregatedData.foreachPartition {
            rows =>
               writePartitionToRedis(rows, redisConfig)
        }

       aggregatedData.write.parquet(s"/data/output.parquet")
    }
}

与我的直觉相反，Spark 调度程序为每个数据接收器（Redis、HDFS/Parquet）生成两个作业。问题是第二个工作也是执行 hive 查询并使工作加倍。我假设两个写操作都会共享来自aggregatedData 阶段的数据。有什么问题还是预期的行为？

Answer 1

您错过了火花的一个基本概念：懒惰。

RDD 不包含任何数据，它只是一组指令，当您调用某个操作（例如将数据写入磁盘/hdfs）时将执行这些指令。如果您重用 RDD（或 Dataframe），则没有存储数据，只需存储每次调用操作时都需要评估的指令。

如果您想重用数据而不需要重新评估 RDD，请使用 .cache() 或最好使用 persist。持久化 RDD 允许您存储转换的结果，以便在未来的迭代中不需要重新评估 RDD。