[HADOOP] Apache Spark : 파티션 당 레코드 수 얻기

Apache Spark : 파티션 당 레코드 수 얻기

전체 파티션 번호와 같은 각 파티션에 대한 정보를 어떻게 얻을 수 있는지 확인하고 싶습니다. 콘솔에 기록하거나 인쇄하기 위해 배포 모드를 사용하여 Spark 작업을 원사 클러스터로 제출할 때 드라이버 측의 각 파티션에있는 레코드를 삭제합니다.

해결법

1. ==============================

   1.다음과 같이 파티션 당 레코드 수를 얻을 수 있습니다.

   다음과 같이 파티션 당 레코드 수를 얻을 수 있습니다.

   df
     .rdd
     .mapPartitionsWithIndex{case (i,rows) => Iterator((i,rows.size))}
     .toDF("partition_number","number_of_records")
     .show
   

   그러나 이것은 또한 스파크 작업을 시작합니다 (레코드 수를 얻기 위해 파일을 스파크로 읽어야하기 때문에).

   스파크는 하이브 테이블 통계를 읽을 수도 있지만, 메타 데이터를 표시하는 방법을 모르겠습니다.

2. ==============================

   2.내장 함수를 사용하고 싶습니다. 그것은 얻는만큼 효율적이어야합니다 :

   내장 함수를 사용하고 싶습니다. 그것은 얻는만큼 효율적이어야합니다 :

   import org.apache.spark.sql.functions.spark_partition_id
   
   df.groupBy(spark_partition_id).count
   

3. ==============================

   3.(sparkPartitionId ()는 org.apache.spark.sql.functions에 있습니다)

   (sparkPartitionId ()는 org.apache.spark.sql.functions에 있습니다)

   import org.apache.spark.sql.functions._ 
   
   df.withColumn("partitionId", sparkPartitionId()).groupBy("partitionId").count.show
   

   @Raphael Roth에 의해 언급 된

   mapPartitionsWithIndex는 RDD 기반 접근 방식 이후로 모든 버전의 스파크에서 작동하는 최상의 접근 방식입니다.

4. ==============================

   4.스파크 / 규모 :

   스파크 / 규모 :

   val numPartitions = 20000
   val a = sc.parallelize(0 until 1e6.toInt, numPartitions )
   val l = a.glom().map(_.length).collect()  # get length of each partition
   print(l.min, l.max, l.sum/l.length, l.length)  # check if skewed
   

   PySpark :

   num_partitions = 20000
   a = sc.parallelize(range(int(1e6)), num_partitions)
   l = a.glom().map(len).collect()  # get length of each partition
   print(min(l), max(l), sum(l)/len(l), len(l))  # check if skewed
   

   RDD뿐 아니라 데이터 프레임에서도 마찬가지입니다. 위의 코드에 DF.rdd.glom ...을 추가하십시오.

   크레딧 : Mike Dusenberry @ https://issues.apache.org/jira/browse/SPARK-17817

from https://stackoverflow.com/questions/46032320/apache-spark-get-number-of-records-per-partition by cc-by-sa and MIT license