AdsHub Analysis Processor - Details for Query 9

Submitted Time: 2026/09/04 01:55:01
Duration: 19 s
Succeeded Jobs: 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159

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

digraph G { 0 [labelType="html" label="Exchange shuffle records written: 1 local merged chunks fetched: 0 shuffle write time total (min, med, max (stageId: taskId)) 1 ms (0 ms, 0 ms, 1 ms (stage 433.0: task 650)) remote merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 479.0: task 713)) 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 479.0: task 713)) remote merged blocks fetched: 0 records read: 1 local bytes read total (min, med, max (stageId: taskId)) 118.0 B (0.0 B, 0.0 B, 118.0 B (stage 434.0: task 698)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 479.0: task 713)) remote bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 479.0: task 713)) merged fetch fallback count: 0 local blocks read: 1 remote merged chunks fetched: 0 remote blocks read: 0 data size total (min, med, max (stageId: taskId)) 96.0 B (0.0 B, 0.0 B, 96.0 B (stage 433.0: task 650)) local merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 479.0: task 713)) number of partitions: 16 remote reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 479.0: task 713)) remote bytes read to disk total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 479.0: task 713)) shuffle bytes written total (min, med, max (stageId: taskId)) 118.0 B (0.0 B, 0.0 B, 118.0 B (stage 433.0: task 650))"]; 1 [labelType="html" label="ObjectHashAggregate number of output rows: 1 time in aggregation build total (min, med, max (stageId: taskId)) 8.6 s (107 ms, 178 ms, 296 ms (stage 433.0: task 670)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 433.0: task 648)) 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 433.0: task 648)) number of total state rows: 2 number of state store instances: 50 memory used by state total (min, med, max (stageId: taskId)) 21.7 KiB (432.0 B, 432.0 B, 784.0 B (stage 433.0: task 676)) count of cache hit on states cache in provider: 300 number of output rows: 1 estimated size of state only on current version total (min, med, max (stageId: taskId)) 5.7 KiB (104.0 B, 104.0 B, 424.0 B (stage 433.0: task 676)) 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 433.0: task 648)) count of cache miss on states cache in provider: 0 time to commit changes total (min, med, max (stageId: taskId)) 8.4 s (107 ms, 177 ms, 295 ms (stage 433.0: task 670)) time to remove total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 433.0: task 648)) number of updated state rows: 1 time to update total (min, med, max (stageId: taskId)) 104 ms (0 ms, 0 ms, 104 ms (stage 433.0: task 650)) number of output rows: 0"]; 3 [labelType="html" label="ObjectHashAggregate number of output rows: 1 time in aggregation build total (min, med, max (stageId: taskId)) 102 ms (0 ms, 0 ms, 102 ms (stage 433.0: task 650)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 433.0: task 648)) number of sort fallback tasks: 0"]; 4 [labelType="html" label="StateStoreRestore number of output rows: 1"]; 5 [labelType="html" label="ObjectHashAggregate number of output rows: 1 time in aggregation build total (min, med, max (stageId: taskId)) 102 ms (0 ms, 0 ms, 102 ms (stage 433.0: task 650)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 433.0: task 648)) number of sort fallback tasks: 0"]; 6 [labelType="html" label="Exchange shuffle records written: 17 local merged chunks fetched: 0 shuffle write time total (min, med, max (stageId: taskId)) 140 ms (0 ms, 0 ms, 83 ms (stage 432.0: task 647)) remote merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 433.0: task 648)) 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 433.0: task 648)) remote merged blocks fetched: 0 records read: 17 local bytes read total (min, med, max (stageId: taskId)) 2.2 KiB (0.0 B, 0.0 B, 2.2 KiB (stage 433.0: task 650)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 433.0: task 648)) remote bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 433.0: task 648)) merged fetch fallback count: 0 local blocks read: 17 remote merged chunks fetched: 0 remote blocks read: 0 data size total (min, med, max (stageId: taskId)) 2.3 KiB (0.0 B, 0.0 B, 136.0 B (stage 432.0: task 631)) local merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 433.0: task 648)) number of partitions: 50 remote reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 433.0: task 648)) remote bytes read to disk total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 433.0: task 648)) shuffle bytes written total (min, med, max (stageId: taskId)) 2.2 KiB (0.0 B, 0.0 B, 136.0 B (stage 432.0: task 634))"]; 7 [labelType="html" label="ObjectHashAggregate number of output rows: 17 time in aggregation build total (min, med, max (stageId: taskId)) 2.8 s (0 ms, 0 ms, 729 ms (stage 432.0: task 631)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 433.0: task 648)) number of sort fallback tasks: 0"]; subgraph cluster8 { isCluster="true"; label="WholeStageCodegen (2)\n \nduration: total (min, med, max (stageId: taskId))\n2.8 s (0 ms, 0 ms, 729 ms (stage 432.0: task 631))"; 9 [labelType="html" label=" Project "]; } 10 [labelType="html" label=" EventTimeWatermark "]; subgraph cluster11 { isCluster="true"; label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n2.9 s (0 ms, 0 ms, 730 ms (stage 432.0: task 631))"; 12 [labelType="html" label=" Project "]; } 13 [labelType="html" label=" Project "]; 14 [labelType="html" label="Filter number of output rows: 17"]; 15 [labelType="html" label="MicroBatchScan number of output rows: 17 estimated number of fetched offsets out of range: 0 number of data loss error: 0"]; 1->0; 2->1; 3->2; 4->3; 5->4; 6->5; 7->6; 9->7; 10->9; 12->10; 13->12; 14->13; 15->14; }

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

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/2026_07_01/referrals_watermark3/state, runId = 906fc2c7-4e02-4dd8-b6bd-e1479cd0eeb1, opId = 0, ver = 3, numPartitions = 50], Update, 1783011783000, 1783011783000, 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/2026_07_01/referrals_watermark3/state, runId = 906fc2c7-4e02-4dd8-b6bd-e1479cd0eeb1, opId = 0, ver = 3, numPartitions = 50], 2

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

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 (2)

EventTimeWatermark created_at#33: timestamp, 15 minutes

Project [data#23.ad_type AS ad_type#26, data#23.created_at AS created_at#33, data#23.reference_link AS reference_link#34, data#23.domain AS domain#40, data#23.date AS date#41]

WholeStageCodegen (1)

Project [from_json(StructField(ip,StringType,true), StructField(ad_type,StringType,true), StructField(bot,IntegerType,true), StructField(title,StringType,true), StructField(url,StringType,true), StructField(uuid,StringType,true), StructField(keyword,StringType,true), StructField(inventory_code,IntegerType,true), StructField(created_at,TimestampType,true), StructField(reference_link,StringType,true), StructField(agent,StringType,true), StructField(referral_type,StringType,true), StructField(tracking_code,StringType,true), StructField(sub_id,StringType,true), StructField(click_type,StringType,true), StructField(domain,StringType,true), StructField(date,StringType,true), StructField(hour,IntegerType,true), StructField(inventory_ad_type,IntegerType,true), StructField(inventory_type_id,IntegerType,true), StructField(is_except_uuid,IntegerType,true), StructField(is_block_uuid,IntegerType,true), StructField(is_except_ip,IntegerType,true), StructField(is_block_ip,IntegerType,true), ... 8 more fields) AS data#23]

Filter ((((((isnotnull(value#8) AND NOT (RLIKE(from_json(StructField(agent,StringType,true), cast(value#8 as string), Some(Etc/UTC)).agent, Yeti|compatible|googlebot|google\.com\/bot\.html) <=> true)) AND (cast(from_json(StructField(date,StringType,true), cast(value#8 as string), Some(Etc/UTC)).date as date) = cast(from_utc_timestamp(2026-09-04 01:55:00.02, Asia/Seoul) as date))) AND isnotnull(regexp_extract(from_json(StructField(reference_link,StringType,true), cast(value#8 as string), Some(Etc/UTC)).reference_link, ^(?:https?:\/\/)?(?:www\.)?([^\/:]+), 1))) AND NOT RLIKE(regexp_extract(from_json(StructField(reference_link,StringType,true), cast(value#8 as string), Some(Etc/UTC)).reference_link, ^(?:https?:\/\/)?(?:www\.)?([^\/:]+), 1), .*\.googlesyndication\.com$)) AND isnotnull(from_json(StructField(domain,StringType,true), cast(value#8 as string), Some(Etc/UTC)).domain)) AND NOT RLIKE(from_json(StructField(domain,StringType,true), cast(value#8 as string), Some(Etc/UTC)).domain, .*\.googlesyndication\.com$))

MicroBatchScan[key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13] class org.apache.spark.sql.kafka010.KafkaSourceProvider$KafkaScan

Details

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


(1) MicroBatchScan
Output [7]: [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13]
class org.apache.spark.sql.kafka010.KafkaSourceProvider$KafkaScan

(2) Filter
Input [7]: [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13]
Condition : ((((((isnotnull(value#8) AND NOT (RLIKE(from_json(StructField(agent,StringType,true), cast(value#8 as string), Some(Etc/UTC)).agent, Yeti|compatible|googlebot|google\.com\/bot\.html) <=> true)) AND (cast(from_json(StructField(date,StringType,true), cast(value#8 as string), Some(Etc/UTC)).date as date) = cast(from_utc_timestamp(2026-09-04 01:55:00.02, Asia/Seoul) as date))) AND isnotnull(regexp_extract(from_json(StructField(reference_link,StringType,true), cast(value#8 as string), Some(Etc/UTC)).reference_link, ^(?:https?:\/\/)?(?:www\.)?([^\/:]+), 1))) AND NOT RLIKE(regexp_extract(from_json(StructField(reference_link,StringType,true), cast(value#8 as string), Some(Etc/UTC)).reference_link, ^(?:https?:\/\/)?(?:www\.)?([^\/:]+), 1), .*\.googlesyndication\.com$)) AND isnotnull(from_json(StructField(domain,StringType,true), cast(value#8 as string), Some(Etc/UTC)).domain)) AND NOT RLIKE(from_json(StructField(domain,StringType,true), cast(value#8 as string), Some(Etc/UTC)).domain, .*\.googlesyndication\.com$))

(3) Project
Output [1]: [from_json(StructField(ip,StringType,true), StructField(ad_type,StringType,true), StructField(bot,IntegerType,true), StructField(title,StringType,true), StructField(url,StringType,true), StructField(uuid,StringType,true), StructField(keyword,StringType,true), StructField(inventory_code,IntegerType,true), StructField(created_at,TimestampType,true), StructField(reference_link,StringType,true), StructField(agent,StringType,true), StructField(referral_type,StringType,true), StructField(tracking_code,StringType,true), StructField(sub_id,StringType,true), StructField(click_type,StringType,true), StructField(domain,StringType,true), StructField(date,StringType,true), StructField(hour,IntegerType,true), StructField(inventory_ad_type,IntegerType,true), StructField(inventory_type_id,IntegerType,true), StructField(is_except_uuid,IntegerType,true), StructField(is_block_uuid,IntegerType,true), StructField(is_except_ip,IntegerType,true), StructField(is_block_ip,IntegerType,true), ... 8 more fields) AS data#23]
Input [7]: [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13]

(4) Project [codegen id : 1]
Output [5]: [data#23.ad_type AS ad_type#26, data#23.created_at AS created_at#33, data#23.reference_link AS reference_link#34, data#23.domain AS domain#40, data#23.date AS date#41]
Input [1]: [data#23]

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

(6) Project [codegen id : 2]
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]

(7) 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#1197L, sum#1199L, sum#1201L, buf#1203]

(8) Exchange
Input [6]: [host#717, date#41, sum#1197L, sum#1199L, sum#1201L, buf#1203]
Arguments: hashpartitioning(host#717, date#41, 50), ENSURE_REQUIREMENTS, [plan_id=2418]

(9) ObjectHashAggregate
Input [6]: [host#717, date#41, sum#1197L, sum#1199L, sum#1201L, 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#1197L, sum#1199L, sum#1201L, buf#1203]

(10) StateStoreRestore
Input [6]: [host#717, date#41, sum#1197L, sum#1199L, sum#1201L, buf#1203]
Arguments: [host#717, date#41], state info [ checkpoint = <unknown>, runId = 609b6e81-3417-441c-9f4d-9302f9b7f90d, opId = 0, ver = 0, numPartitions = 50], 2

(11) ObjectHashAggregate
Input [6]: [host#717, date#41, sum#1197L, sum#1199L, sum#1201L, 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#1197L, sum#1199L, sum#1201L, buf#1203]

(12) StateStoreSave
Input [6]: [host#717, date#41, sum#1197L, sum#1199L, sum#1201L, buf#1203]
Arguments: [host#717, date#41], state info [ checkpoint = <unknown>, runId = 609b6e81-3417-441c-9f4d-9302f9b7f90d, opId = 0, ver = 0, numPartitions = 50], Append, -9223372036854775808, -9223372036854775808, 2

(13) ObjectHashAggregate
Input [6]: [host#717, date#41, sum#1197L, sum#1199L, sum#1201L, 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]

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

SQL / DataFrame Properties

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