[ https://issues.apache.org/jira/browse/IMPALA-7751?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16665425#comment-16665425 ]
Lars Volker commented on IMPALA-7751: ------------------------------------- I see. Impala would need to detect that it pruned all but one partitions on the *sink* side of the table and thus remove them from the partitioning criteria for the pre-insert shuffle. We currently perform partition pruning on the *scans* and I suspect we'd need new logic to apply that to the sinks. It seems like a substantial change to me, but [~twmarshall] has a better insight into the Kudu integration and might have more ideas. > Kudu insert statement should push down range partition predicates > ----------------------------------------------------------------- > > Key: IMPALA-7751 > URL: https://issues.apache.org/jira/browse/IMPALA-7751 > Project: IMPALA > Issue Type: Improvement > Reporter: Quanlong Huang > Priority: Major > Attachments: metrics1.tsv, metrics2.tsv, metrics3.tsv, profile.txt > > > We have a job dumping newly added data in HDFS into Kudu table for good > performance of point queries. Each day we create a new range partition in > Kudu for the new data on this day. When we add more and more Kudu range > partitions, we found performance degradation of this job. > The root cause is, the insert statement for kudu does not leverage the > partition predicates for kudu range partition keys, which causes skew on the > insert nodes. > How to reveal this: > Step 1: Launch impala cluster with 3 nodes. > Step 2: Create an HDFS table with more than 3 underlying files, thus will > have more than 3 scan ranges > {code:sql} > create table default.metrics_tbl ( > source_id string, > event_timestamp bigint, > value double > ) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE; > {code} > Upload the three attached tsv files into its directory and refresh this table > in Impala. > Step 3: Create a Kudu table with mix partitions containing 3 hash partitions > and 3 range partitions. > {code:sql} > create table default.metrics_kudu_tbl ( > source_id string, > event_timestamp bigint, > value double, > primary key(source_id, event_timestamp) > ) partition by > hash (source_id) PARTITIONS 3, > range (event_timestamp) ( > partition 0 <= values < 10000, > partition 10000 <= values < 20000, > partition 20000 <= values < 30000 > ) stored as kudu; > {code} > Step 4: Dump rows in HDFS table into Kudu giving partition predicates. > {code:sql} > insert into table metrics_kudu_tbl > select source_id, event_timestamp, value from metrics_tbl > where event_timestamp >= 10000 and event_timestamp < 20000; > {code} > Step 5: Looking into the profile, there're three fragment instances > containing KuduTableSink but only one of them received and generated data. > {code:java} > Averaged Fragment F01: > KuduTableSink: > - TotalNumRows: 1.00K (1000) > Fragment F01: > Instance 6347506799a2966d:6e82f49200000004 > KuduTableSink: > - TotalNumRows: 3.00K (3000) > Instance 6347506799a2966d:6e82f49200000005 > KuduTableSink: > - TotalNumRows: 0 (0) > Instance 6347506799a2966d:6e82f49200000003 > KuduTableSink: > - TotalNumRows: 0 (0) > {code} > Thus, only one fragment instance of F01 is sorting and ingesting data into > Impala. > Generally, if there're N range partitions and all the inserted rows are > belong to one range (supplied by the partition predicates in WHERE clause), > only 1/N of the insert fragments are producing data. -- This message was sent by Atlassian JIRA (v7.6.3#76005) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-all-unsubscr...@impala.apache.org For additional commands, e-mail: issues-all-h...@impala.apache.org