쓰기 및 다시 읽을 때 스파크가 누락 됨

Question

데이터 집합을 만들고이 파일을 s3에 ${BasePath}/y=2107/m=11/d=16/쪽 마루 형식으로 쓰려고합니다. 오늘 날짜의 데이터를 작성한 다음 s3에 기록한 후 지난 10 일 동안 데이터를 다시 읽어야합니다.

dataset.write.mode(SaveMode.Overwrite) 
       .format(sourceConfig.format).save(getWriteBasePath(sourceConfig.sourcePath 
       , replaceDate)) 
newDf = spark.read.parquet("path1",...,"path9").union(dataset) 

두번째 접근법처럼, 9 일 개 데이터의 나머지 부분을 읽어와 노동 조합을, S3에 현재 날짜 데이터 쓰기 : 쓰기

나는이 작업을 수행 할 수있는 2 가지 방법,

1 접근입니다 현재 날짜 데이터를 s3으로 변경하고 모든 10 일 데이터를 다시 읽습니다. 같은 뭔가 :

 dataset.write.mode(SaveMode.Overwrite) 
      .format(sourceConfig.format).save(getWriteBasePath(sourceConfig.sourcePath 
      , replaceDate)) 
    newDf = spark.read.parquet("path1",...,"path10") 

첫번째 접근 방식은 문제없이 작업 하지만 두번째 방법에 비해 매우 느립니다. 하지만 두 번째 방법을 사용하면 전체 데이터를 다시 읽은 후 현재 열의 일부가 null 값을 얻게됩니다. 나는 기록 된 당일의 자료가 적절하다는 것을 확인했다.

왜 이런 일이 발생하는지 알 수 없습니다. 나는 다음과 같은 특성을 가진 스파크 컨텍스트를 만드는 오전 :

sparkSession.sparkContext.hadoopConfiguration.set("mapreduce.fileoutputcommitter.algorithm.version", "2") 
     sparkSession.sparkContext.getConf.set("spark.hadoop.parquet.enable.summary-metadata", "false") 
     sparkSession.sparkContext.getConf.set("spark.sql.parquet.mergeSchema", "false") 
     sparkSession.sparkContext.getConf.set("spark.sql.parquet.filterPushdown", "true") 
     sparkSession.sparkContext.getConf.set("spark.sql.hive.metastorePartitionPruning", "true") 

Answer 1

내가 다른 dataframe에 모든 과거 구일 하나 dataframe의 현재 데이터를 캐시 읽기하는 것을 선호합니다. 이제 모든 데이터를 메모리에 저장 했으므로 실행 속도가 빨라야합니다. 우리가 s3에 쓰는 순간 (행동으로) 생성 된 전체 DAG가 플러시되고 첫 번째 방법이 실제로 느린 이유가 그 것이다.