AdsHub Analysis Processor - Details for Query 7

Submitted Time: 2025/12/05 22:18:00
Duration: 3 s
Succeeded Jobs: 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127

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)) 0 ms (0 ms, 0 ms, 0 ms (stage 337.0: task 494)) remote merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 365.0: task 528)) 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 365.0: task 528)) remote merged blocks fetched: 0 records read: 1 local bytes read total (min, med, max (stageId: taskId)) 144.0 B (0.0 B, 0.0 B, 144.0 B (stage 353.0: task 524)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 365.0: task 528)) remote bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 365.0: task 528)) 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)) 128.0 B (0.0 B, 0.0 B, 128.0 B (stage 337.0: task 494)) local merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 365.0: task 528)) number of partitions: 16 remote reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 365.0: task 528)) remote bytes read to disk total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 365.0: task 528)) shuffle bytes written total (min, med, max (stageId: taskId)) 144.0 B (0.0 B, 0.0 B, 144.0 B (stage 337.0: task 494))"]; 1 [labelType="html" label="ObjectHashAggregate number of output rows: 1 time in aggregation build total (min, med, max (stageId: taskId)) 8.3 s (127 ms, 159 ms, 234 ms (stage 337.0: task 495)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 337.0: task 469)) 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 337.0: task 469)) number of total state rows: 1 number of state store instances: 50 memory used by state total (min, med, max (stageId: taskId)) 21.6 KiB (432.0 B, 432.0 B, 968.0 B (stage 337.0: task 494)) count of cache hit on states cache in provider: 200 number of output rows: 1 estimated size of state only on current version total (min, med, max (stageId: taskId)) 5.4 KiB (104.0 B, 104.0 B, 448.0 B (stage 337.0: task 494)) 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 337.0: task 469)) count of cache miss on states cache in provider: 0 time to commit changes total (min, med, max (stageId: taskId)) 8.1 s (81 ms, 158 ms, 233 ms (stage 337.0: task 495)) time to remove total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 337.0: task 469)) number of updated state rows: 1 time to update total (min, med, max (stageId: taskId)) 116 ms (0 ms, 0 ms, 116 ms (stage 337.0: task 494)) 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)) 27 ms (0 ms, 0 ms, 27 ms (stage 337.0: task 494)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 337.0: task 469)) number of sort fallback tasks: 0"]; 4 [labelType="html" label="StateStoreRestore number of output rows: 2"]; 5 [labelType="html" label="ObjectHashAggregate number of output rows: 1 time in aggregation build total (min, med, max (stageId: taskId)) 11 ms (0 ms, 0 ms, 11 ms (stage 337.0: task 494)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 337.0: task 469)) number of sort fallback tasks: 0"]; 6 [labelType="html" label="Exchange shuffle records written: 2 local merged chunks fetched: 0 shuffle write time total (min, med, max (stageId: taskId)) 3 ms (0 ms, 0 ms, 2 ms (stage 336.0: task 468)) remote merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 337.0: task 469)) 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 337.0: task 469)) remote merged blocks fetched: 0 records read: 2 local bytes read total (min, med, max (stageId: taskId)) 254.0 B (0.0 B, 0.0 B, 254.0 B (stage 337.0: task 494)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 337.0: task 469)) remote bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 337.0: task 469)) merged fetch fallback count: 0 local blocks read: 2 remote merged chunks fetched: 0 remote blocks read: 0 data size total (min, med, max (stageId: taskId)) 264.0 B (0.0 B, 0.0 B, 136.0 B (stage 336.0: task 467)) local merged bytes read total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 337.0: task 469)) number of partitions: 50 remote reqs duration total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 337.0: task 469)) remote bytes read to disk total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 337.0: task 469)) shuffle bytes written total (min, med, max (stageId: taskId)) 254.0 B (0.0 B, 0.0 B, 132.0 B (stage 336.0: task 467))"]; 7 [labelType="html" label="ObjectHashAggregate number of output rows: 2 time in aggregation build total (min, med, max (stageId: taskId)) 1.2 s (0 ms, 0 ms, 591 ms (stage 336.0: task 467)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 336.0: task 467)) number of sort fallback tasks: 0"]; subgraph cluster8 { isCluster="true"; label="WholeStageCodegen (2)\n \nduration: total (min, med, max (stageId: taskId))\n1.2 s (0 ms, 0 ms, 591 ms (stage 336.0: task 467))"; 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))\n1.2 s (0 ms, 0 ms, 592 ms (stage 336.0: task 468))"; 12 [labelType="html" label=" Project "]; } 13 [labelType="html" label=" Project "]; 14 [labelType="html" label="Filter number of output rows: 2"]; 15 [labelType="html" label="MicroBatchScan number of output rows: 2 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=1909]

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 = 2, numPartitions = 50], Update, 0, 1765035446000, 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 = 2, numPartitions = 50], 2

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

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 22:18:00.014, 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 22:18:00.014, 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=2012]

(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 = d7d1f5e2-ba0f-4e20-89e3-3e36f3e0fe5c, 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 = d7d1f5e2-ba0f-4e20-89e3-3e36f3e0fe5c, 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=2018]

SQL / DataFrame Properties

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