Spark数据集选择性重新计算

Question

我知道Spark知道如何在新节点上分配需要完成的工作，例如，在另一个节点出现故障时启动新节点。

我想知道这是否可以用于其他用例。

假设我有一个转换和操作树。当其中一个数据集/数据帧被更新时会发生什么(例如，导入了新文件)。在这种情况下，我只想重复那些受到影响并与此更改相关的转换和操作。其他不相关的转换和操作应该从缓存中使用，因为它们没有受到影响。

现在，如果我只有这些数据帧、转换和操作中的一些，我可以手动完成。但我有几十个或更多这样的DF和行动，我正在努力了解spark在框架内是否有什么东西可以帮助我。

下面是我的代码示例：

val carLines = spark
.read
.option("header", "true")
.schema(carLineSchema)
.csv("src/test/resources/cars")
val ageMappingFunction: Int => String = (age: Int) => if (age > 80) "old" else "young"
//
val _age = udf.register("_age", ageMappingFunction)
val personLines = spark
.read
.option("header", "true")
.schema(personLineSchema)
.csv("src/test/resources/persons")
.withColumn("_age", _age($"age"))
val accidentsLines = spark
.read
.option("header", "true")
.schema(accidentLineSchema)
.csv("src/test/resources/accidents")
val carOwners = personLines
.withColumnRenamed("id", "driver_id")
.join(carLines, Seq("driver_id"), "left")
.withColumnRenamed("id", "car_id")
.withColumnRenamed("car_make", "car_maker")
.withColumnRenamed("driver_id", "id")

现在进行一些转换：

val accidentsWithDrivers = accidentsLines
.join(personLines.withColumnRenamed("id", "driver_id"), "driver_id")
val accidentsPerDriverID = accidentsWithDrivers
.groupBy("driver_id")
.agg(Map(
"name" -> "count"
))
.withColumnRenamed("count(name)", "accident_count")
.withColumnRenamed("driver_id", "id")
val finalTable = carOwners
.join(numberOfCarsPerDriver, Seq("id", "name", "age", "_age"))
.join(accidentsPerDriverID, "id")

然后我做一些操作(为了简单起见，我将使用"show")：

carOwners.show(true)
numberOfCarsPerDriver.show(true)
finalTable.show(true)

所以，我想问的是，如果accidentsLines发生了变化，但carLines或personLines没有变化，会怎么样。我们可以用缓存的carLines和personLines值进行carOwners转换吗？

换句话说：假设我想把RDD#cache()api保存在spark集群的内存中，我能以某种方式使用它在不同的驱动程序运行之间生存吗？

Answer 1

我需要使用作业服务器或使用Apache Ignite:的IgniteRDD支持

//WRITE
val igniteContext = new IgniteContext(spark.sparkContext, "ignite-config.xml", true)
val schema = dataframe.schema
val rdd = dataframe.rdd
igniteContext.ignite().getOrCreateCache("ignite-cache").put("schema", schema)
igniteContext.fromCache(name).saveValues(rdd)
//READ
val schema = igniteContext.ignite()
.getOrCreateCache[String, StructType]("ignite-cache")
.get("schema")
.asInstanceOf[StructType]
val igniteRdd: IgniteRDD[String, Row] = igniteContext.fromCache(name)
val rdd = igniteRdd.map(a => a._2)
val dataframe = spark.createDataFrame(rdd, schema)