不一致计数后窗口引导功能，和过滤器

Question

编辑2:

我已经报告了这作为一个问题火花开发人员，我会在这里发布状态当我得到一些。

我有一个问题已经困扰我很长一段时间了。

null

import org.apache.spark.sql.expressions.Window
val w = Window.partitionBy("user_id").orderBy("start")
val ts_lead = coalesce(lead("start", 1) .over(w), date_add(col("start"), 1))

val df2 = df1.
    withColumn("end", ts_lead).
    withColumn("duration", col("end").cast("long")-col("start").cast("long"))

df2.where("type='B' and duration>4").count()

• 运行1:19359949
• 运行2:19359964

如果我单独运行每个过滤器，一切都很好，我得到一致的结果。但如果我把它们结合起来，就不一致了。

我试着过滤来分离数据流，先持续时间再类型，反之亦然，也没有乐趣。我知道我可以缓存或检查点datframe，但它是非常大的dataset，而且我多次进行类似的计算，所以我不能真正为检查点和缓存腾出时间和磁盘空间。

import org.apache.spark.sql.expressions.Window

val getRandomUser = udf(()=>{
    val users = Seq("John","Eve","Anna","Martin","Joe","Steve","Katy")
   users(scala.util.Random.nextInt(7))
})

val getRandomType = udf(()=>{
    val types = Seq("TypeA","TypeB","TypeC","TypeD","TypeE")
    types(scala.util.Random.nextInt(5))
})

val getRandomStart = udf((x:Int)=>{
    x+scala.util.Random.nextInt(47)
})

for( a <- 0 to 23){
        // use iterator a to continue with next million, repeat 1 mil times
        val x=Range(a*1000000,(a*1000000)+1000000).toDF("id").
            withColumn("start",getRandomStart(col("id"))).
            withColumn("user",getRandomUser()).
            withColumn("type",getRandomType()).
            drop("id")

        x.write.mode("append").orc("s3://your-bucket/random.orc")
}

val w = Window.partitionBy("user").orderBy("start")
val ts_lead = coalesce(lead("start", 1) .over(w), lit(30000000))

val fox2 = spark.read.orc("s3://your-bucket/random.orc").
    withColumn("end", ts_lead).
    withColumn("duration", col("end")-col("start"))

// repeated executions of this line returns different results for count 
fox2.where("type='TypeA' and duration>4").count()

• 运行1:2551259
• 运行2:2550756
• 运行3:2551279

每次运行不同计数

Answer 1

我已经在当地复制了你的问题。

据我所知，问题在于您在这句话中是按duration进行筛选:fox2.where（“type='typea'and duration>4”）.count()

并且持续时间是随机生成的。我知道您使用的是种子，但如果将其并行化，则不知道将向每个ID添加哪个随机值。

1 + 21
2 + 14
3 + 5
4 + 17

1 + 21
4 + 14
3 + 5
2 + 17