1
저는 Spark-2.2를 사용하고 있습니다. 나는 스파크의 버킷을 Pocing하고 있습니다.스파크 버킷 팅 및 collect_set 집계
내가이 요청에 의한 그룹의 설명 실행하고있어+--------------------+--------------------+-------+
| col_name| data_type|comment|
+--------------------+--------------------+-------+
| bundle| string| null|
| ifa| string| null|
| date_| date| null|
| hour| int| null|
| | | |
|# Detailed Table ...| | |
| Database| default| |
| Table| my_bucketed_tbl|
| Owner| zeppelin| |
| Created|Thu Dec 21 13:43:...| |
| Last Access|Thu Jan 01 00:00:...| |
| Type| EXTERNAL| |
| Provider| orc| |
| Num Buckets| 16| |
| Bucket Columns| [`ifa`]| |
| Sort Columns| [`ifa`]| |
| Table Properties|[transient_lastDd...| |
| Location|hdfs:/user/hive/w...| |
| Serde Library|org.apache.hadoop...| |
| InputFormat|org.apache.hadoop...| |
| OutputFormat|org.apache.hadoop...| |
| Storage Properties|[serialization.fo...| |
+--------------------+--------------------+-------+
, 내가 볼 수있는 우리가 교환 단계 아끼지했다고 :
sql("select ifa,max(bundle) from my_bucketed_tbl group by ifa").explain
== Physical Plan ==
SortAggregate(key=[ifa#932], functions=[max(bundle#920)])
+- SortAggregate(key=[ifa#932], functions=[partial_max(bundle#920)])
+- *Sort [ifa#932 ASC NULLS FIRST], false, 0
+- *FileScan orc default.level_1[bundle#920,ifa#932] Batched: false, Format: ORC, Location: InMemoryFileIndex[hdfs://ip-10-44-9-73.ec2.internal:8020/user/hive/warehouse/level_1/date_=2017-1..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<bundle:string,ifa:string>
을 나는 bucketed 테이블을 만들었습니다, 여기에
desc formatted my_bucketed_tbl
출력입니다
하지만, 내가 collect_set
와 함께 불꽃의 max
기능을 교체 할 때 내가 실행 계획은 교환 단계가 절약되지 않고, 비 bucketed 테이블과 같은 의미 있다고 볼 수
sql("select ifa,collect_set(bundle) from my_bucketed_tbl group by ifa").explain
== Physical Plan ==
ObjectHashAggregate(keys=[ifa#1010], functions=[collect_set(bundle#998, 0, 0)])
+- Exchange hashpartitioning(ifa#1010, 200)
+- ObjectHashAggregate(keys=[ifa#1010], functions=[partial_collect_set(bundle#998, 0, 0)])
+- *FileScan orc default.level_1[bundle#998,ifa#1010] Batched: false, Format: ORC, Location: InMemoryFileIndex[hdfs://ip-10-44-9-73.ec2.internal:8020/user/hive/warehouse/level_1/date_=2017-1..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<bundle:string,ifa:string>
내가 놓친 구성이 있습니까? 아니면 Spark의 버킷이 현재 가지고있는 제한 사항입니까?
관리되는 테이블에서 잘 작동합니다. – user6910411