AdsHub Analysis Processor - Details for Query 44

Submitted Time: 2025/12/06 04:49:00
Duration: 2 s
Succeeded Jobs: 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719

Show the Stage ID and Task ID that corresponds to the max metric

digraph G { 0 [labelType="html" label="Exchange shuffle records written: 0 local merged chunks fetched: 0 shuffle write time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2117.0: task 3015)) remote merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2117.0: task 3015)) local merged blocks fetched: 0 corrupt merged block chunks: 0 remote merged reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2117.0: task 3015)) remote merged blocks fetched: 0 records read: 0 local bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2117.0: task 3015)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2117.0: task 3015)) remote bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2117.0: task 3015)) merged fetch fallback count: 0 local blocks read: 0 remote merged chunks fetched: 0 remote blocks read: 0 data size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2117.0: task 3015)) local merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2117.0: task 3015)) number of partitions: 16 remote reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2117.0: task 3015)) remote bytes read to disk total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2117.0: task 3015)) shuffle bytes written total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2117.0: task 3015))"]; 1 [labelType="html" label="ObjectHashAggregate number of output rows: 0 time in aggregation build total (min, med, max (stageId: taskId)) 6.7 s (114 ms, 133 ms, 167 ms (stage 2113.0: task 2973)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) number of sort fallback tasks: 0"]; 2 [labelType="html" label="StateStoreSave number of shuffle partitions: 50 number of removed state rows: 0 data returned from Python workers total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) number of total state rows: 7 number of state store instances: 50 memory used by state total (min, med, max (stageId: taskId)) 23.5 KiB (432.0 B, 432.0 B, 856.0 B (stage 2113.0: task 2989)) count of cache hit on states cache in provider: 1,500 number of output rows: 0 estimated size of state only on current version total (min, med, max (stageId: taskId)) 7.2 KiB (104.0 B, 104.0 B, 496.0 B (stage 2113.0: task 2989)) number of rows which are dropped by watermark: 0 data sent to Python workers total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) count of cache miss on states cache in provider: 0 time to commit changes total (min, med, max (stageId: taskId)) 6.7 s (113 ms, 133 ms, 167 ms (stage 2113.0: task 2973)) time to remove total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) number of updated state rows: 0 time to update total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) number of output rows: 0"]; 3 [labelType="html" label="ObjectHashAggregate number of output rows: 0 time in aggregation build total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) number of sort fallback tasks: 0"]; 4 [labelType="html" label="StateStoreRestore number of output rows: 0"]; 5 [labelType="html" label="ObjectHashAggregate number of output rows: 0 time in aggregation build total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) number of sort fallback tasks: 0"]; 6 [labelType="html" label="Exchange shuffle records written: 0 local merged chunks fetched: 0 shuffle write time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) remote merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) local merged blocks fetched: 0 corrupt merged block chunks: 0 remote merged reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) remote merged blocks fetched: 0 records read: 0 local bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) remote bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) merged fetch fallback count: 0 local blocks read: 0 remote merged chunks fetched: 0 remote blocks read: 0 data size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) local merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) number of partitions: 50 remote reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) remote bytes read to disk total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) shuffle bytes written total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964))"]; 7 [labelType="html" label="ObjectHashAggregate number of output rows: 0 time in aggregation build total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 2113.0: task 2964)) number of sort fallback tasks: 0"]; subgraph cluster8 { isCluster="true"; label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n0 ms (0 ms, 0 ms, 0 ms (stage 2113.0: task 2964))"; 9 [labelType="html" label=" Project "]; } 10 [labelType="html" label=" EventTimeWatermark "]; 11 [labelType="html" label="LocalTableScan number of output rows: 0"]; 1->0; 2->1; 3->2; 4->3; 5->4; 6->5; 7->6; 9->7; 10->9; 11->10; }

Exchange RoundRobinPartitioning(16), REPARTITION_BY_NUM, [plan_id=11241]

ObjectHashAggregate(keys=[host#717, date#41], functions=[sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END), sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END), sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END), collect_set(domain#40, 0, 0)])

StateStoreSave [host#717, date#41], state info [ checkpoint = file:/app/checkpoint/2025_12_06/referrals_watermark3/state, runId = 184404a1-7672-47f3-81a6-842372b55bc2, opId = 0, ver = 15, numPartitions = 50], Update, 1765052727000, 1765058600000, 2

ObjectHashAggregate(keys=[host#717, date#41], functions=[merge_sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END), merge_sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END), merge_sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END), merge_collect_set(domain#40, 0, 0)])

StateStoreRestore [host#717, date#41], state info [ checkpoint = file:/app/checkpoint/2025_12_06/referrals_watermark3/state, runId = 184404a1-7672-47f3-81a6-842372b55bc2, opId = 0, ver = 15, numPartitions = 50], 2

Exchange hashpartitioning(host#717, date#41, 50), ENSURE_REQUIREMENTS, [plan_id=11235]

ObjectHashAggregate(keys=[host#717, date#41], functions=[partial_sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END), partial_sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END), partial_sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END), partial_collect_set(domain#40, 0, 0)])

Project [ad_type#26, domain#40, date#41, regexp_extract(reference_link#34, ^(?:https?:\/\/)?(?:www\.)?([^\/:]+), 1) AS host#717]

WholeStageCodegen (1)

EventTimeWatermark created_at#33: timestamp, 15 minutes

LocalTableScan <empty>, [ad_type#26, created_at#33, reference_link#34, domain#40, date#41]

Details

== Physical Plan ==
Exchange (11)
+- ObjectHashAggregate (10)
   +- StateStoreSave (9)
      +- ObjectHashAggregate (8)
         +- StateStoreRestore (7)
            +- ObjectHashAggregate (6)
               +- Exchange (5)
                  +- ObjectHashAggregate (4)
                     +- * Project (3)
                        +- EventTimeWatermark (2)
                           +- LocalTableScan (1)


(1) LocalTableScan
Output [5]: [ad_type#26, created_at#33, reference_link#34, domain#40, date#41]
Arguments: <empty>, [ad_type#26, created_at#33, reference_link#34, domain#40, date#41]

(2) EventTimeWatermark
Input [5]: [ad_type#26, created_at#33, reference_link#34, domain#40, date#41]
Arguments: created_at#33: timestamp, 15 minutes

(3) Project [codegen id : 1]
Output [4]: [ad_type#26, domain#40, date#41, regexp_extract(reference_link#34, ^(?:https?:\/\/)?(?:www\.)?([^\/:]+), 1) AS host#717]
Input [5]: [ad_type#26, created_at#33-T900000ms, reference_link#34, domain#40, date#41]

(4) ObjectHashAggregate
Input [4]: [ad_type#26, domain#40, date#41, host#717]
Keys [2]: [host#717, date#41]
Functions [4]: [partial_sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END), partial_sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END), partial_sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END), partial_collect_set(domain#40, 0, 0)]
Aggregate Attributes [4]: [sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END)#784L, sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END)#786L, sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END)#788L, collect_set(domain#40, 0, 0)#782]
Results [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]

(5) Exchange
Input [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]
Arguments: hashpartitioning(host#717, date#41, 50), ENSURE_REQUIREMENTS, [plan_id=11325]

(6) ObjectHashAggregate
Input [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]
Keys [2]: [host#717, date#41]
Functions [4]: [merge_sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END), merge_sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END), merge_sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END), merge_collect_set(domain#40, 0, 0)]
Aggregate Attributes [4]: [sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END)#784L, sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END)#786L, sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END)#788L, collect_set(domain#40, 0, 0)#782]
Results [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]

(7) StateStoreRestore
Input [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]
Arguments: [host#717, date#41], state info [ checkpoint = <unknown>, runId = db01ca09-5207-4a42-8d82-af19fc66a8f7, opId = 0, ver = 0, numPartitions = 50], 2

(8) ObjectHashAggregate
Input [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]
Keys [2]: [host#717, date#41]
Functions [4]: [merge_sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END), merge_sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END), merge_sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END), merge_collect_set(domain#40, 0, 0)]
Aggregate Attributes [4]: [sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END)#784L, sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END)#786L, sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END)#788L, collect_set(domain#40, 0, 0)#782]
Results [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]

(9) StateStoreSave
Input [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]
Arguments: [host#717, date#41], state info [ checkpoint = <unknown>, runId = db01ca09-5207-4a42-8d82-af19fc66a8f7, opId = 0, ver = 0, numPartitions = 50], Append, -9223372036854775808, -9223372036854775808, 2

(10) ObjectHashAggregate
Input [6]: [host#717, date#41, sum#1192L, sum#1194L, sum#1197L, buf#1203]
Keys [2]: [host#717, date#41]
Functions [4]: [sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END), sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END), sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END), collect_set(domain#40, 0, 0)]
Aggregate Attributes [4]: [sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END)#784L, sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END)#786L, sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END)#788L, collect_set(domain#40, 0, 0)#782]
Results [6]: [host#717, date#41, sum(CASE WHEN (ad_type#26 = request) THEN 1 ELSE 0 END)#784L AS request#785L, sum(CASE WHEN (ad_type#26 = impression) THEN 1 ELSE 0 END)#786L AS impression#787L, sum(CASE WHEN NOT ad_type#26 IN (impression,request) THEN 1 ELSE 0 END)#788L AS click#789L, concat_ws(,, collect_set(domain#40, 0, 0)#782) AS domains#796]

(11) Exchange
Input [6]: [host#717, date#41, request#785L, impression#787L, click#789L, domains#796]
Arguments: RoundRobinPartitioning(16), REPARTITION_BY_NUM, [plan_id=11331]

SQL / DataFrame Properties

Name	Value
spark.sql.adaptive.enabled	false
spark.sql.cbo.enabled	false
spark.sql.requireAllClusterKeysForDistribution	false