AdsHub Analysis Processor - Details for Query 12

Submitted Time: 2025/12/05 23:04:00
Duration: 3 s
Succeeded Jobs: 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207

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 577.0: task 835)) remote merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 614.0: task 867)) 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 614.0: task 867)) remote merged blocks fetched: 0 records read: 1 local bytes read total (min, med, max (stageId: taskId)) 130.0 B (0.0 B, 0.0 B, 130.0 B (stage 578.0: task 855)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 614.0: task 867)) remote bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 614.0: task 867)) 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)) 104.0 B (0.0 B, 0.0 B, 104.0 B (stage 577.0: task 835)) local merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 614.0: task 867)) number of partitions: 16 remote reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 614.0: task 867)) remote bytes read to disk total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 614.0: task 867)) shuffle bytes written total (min, med, max (stageId: taskId)) 130.0 B (0.0 B, 0.0 B, 130.0 B (stage 577.0: task 835))"]; 1 [labelType="html" label="ObjectHashAggregate number of output rows: 1 time in aggregation build total (min, med, max (stageId: taskId)) 7.3 s (116 ms, 147 ms, 185 ms (stage 577.0: task 824)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 577.0: task 809)) 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 577.0: task 809)) number of total state rows: 3 number of state store instances: 50 memory used by state total (min, med, max (stageId: taskId)) 22.1 KiB (432.0 B, 432.0 B, 808.0 B (stage 577.0: task 830)) count of cache hit on states cache in provider: 400 number of output rows: 1 estimated size of state only on current version total (min, med, max (stageId: taskId)) 6.0 KiB (104.0 B, 104.0 B, 448.0 B (stage 577.0: task 830)) 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 577.0: task 809)) count of cache miss on states cache in provider: 0 time to commit changes total (min, med, max (stageId: taskId)) 7.2 s (87 ms, 147 ms, 184 ms (stage 577.0: task 823)) time to remove total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 577.0: task 809)) number of updated state rows: 1 time to update total (min, med, max (stageId: taskId)) 69 ms (0 ms, 0 ms, 69 ms (stage 577.0: task 835)) 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)) 8 ms (0 ms, 0 ms, 8 ms (stage 577.0: task 835)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 577.0: task 809)) 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)) 14 ms (0 ms, 0 ms, 14 ms (stage 577.0: task 835)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 577.0: task 809)) number of sort fallback tasks: 0"]; 6 [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 576.0: task 804)) remote merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 577.0: task 809)) 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 577.0: task 809)) remote merged blocks fetched: 0 records read: 1 local bytes read total (min, med, max (stageId: taskId)) 146.0 B (0.0 B, 0.0 B, 146.0 B (stage 577.0: task 835)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 577.0: task 809)) remote bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 577.0: task 809)) 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)) 144.0 B (0.0 B, 0.0 B, 144.0 B (stage 576.0: task 804)) local merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 577.0: task 809)) number of partitions: 50 remote reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 577.0: task 809)) remote bytes read to disk total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 577.0: task 809)) shuffle bytes written total (min, med, max (stageId: taskId)) 146.0 B (0.0 B, 0.0 B, 146.0 B (stage 576.0: task 804))"]; 7 [labelType="html" label="ObjectHashAggregate number of output rows: 1 time in aggregation build total (min, med, max (stageId: taskId)) 584 ms (0 ms, 0 ms, 584 ms (stage 576.0: task 804)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 577.0: task 809)) number of sort fallback tasks: 0"]; subgraph cluster8 { isCluster="true"; label="WholeStageCodegen (2)\n \nduration: total (min, med, max (stageId: taskId))\n584 ms (0 ms, 0 ms, 584 ms (stage 576.0: task 804))"; 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))\n584 ms (0 ms, 0 ms, 584 ms (stage 576.0: task 804))"; 12 [labelType="html" label=" Project "]; } 13 [labelType="html" label=" Project "]; 14 [labelType="html" label="Filter number of output rows: 1"]; 15 [labelType="html" label="MicroBatchScan number of output rows: 1 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=3165]

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 = 4, numPartitions = 50], Update, 1765036357000, 1765037036000, 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 = 4, numPartitions = 50], 2

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

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(2025-12-05 23:04:00.022, 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(2025-12-05 23:04:00.022, 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#1192L, sum#1194L, sum#1197L, buf#1203]

(8) 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=3268]

(9) 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]

(10) 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 = 63e580a7-b866-4653-b0e7-58a55f3bc36d, opId = 0, ver = 0, numPartitions = 50], 2

(11) 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]

(12) 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 = 63e580a7-b866-4653-b0e7-58a55f3bc36d, opId = 0, ver = 0, numPartitions = 50], Append, -9223372036854775808, -9223372036854775808, 2

(13) 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]

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

SQL / DataFrame Properties

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