Spark DataFrame 행 개수가 실행 사이에 일치하지 않습니다.

Question

EMR에서 스파크 작업 (버전 2.1.1)을 실행할 때 각 실행은 데이터 프레임에서 다른 양의 행을 계산합니다. 먼저 s3에서 4 개의 다른 데이터 프레임으로 데이터를 읽었습니다.이 개수는 데이터 프레임에 합류 한 후 항상 일치합니다. 조인의 결과는 서로 다릅니다. 이후 나는 또한 결과를 필터링하고 각 실행마다 다른 카운트를 갖는다. 변형은 1 ~ 5 행 차이가 작지만 이해할 수있는 부분입니다.

이

을위한 코드 인 조인

val impJoinKey = Seq("iid", "globalVisitorKey", "date") 

val impressionsJoined: DataFrame = impressionDsNoDuplicates 
    .join(realUrlDSwithDatenoDuplicates, impJoinKey, "outer") 
    .join(impressionParamterDSwithDateNoDuplicates, impJoinKey, "left") 
    .join(chartSiteInstance, impJoinKey, "left") 
    .withColumn("timestamp", coalesce($"timestampImp", $"timestampReal", $"timestampParam")) 
    .withColumn("url", coalesce($"realUrl", $"url")) 

하고 이것은 필터이다

val impressionsJoined: Dataset[ImpressionJoined] = impressionsJoinedFullDay.where($"timestamp".geq(new Timestamp(start.getMillis))).cache() 

I 대신 여기서하지만 동일한 결과

와의 필터 방법을 사용하여 시도

아무 생각하세요?

감사 Nir 씨

Answer 1

는 데이터 소스 중 하나는 시간이 지남에 따라 변화 가능성이있다?
impressionsJoined은 캐시되지 않으므로 spark는 모든 작업에 대해 처음부터 다시 평가하고 소스에서 데이터를 다시 읽는 것을 포함합니다.

조인 후에 impressionsJoined 캐싱을 시도하십시오.