I85PREP
⌘ K
Cloudera CDP Data Analyst study materials
Cloudera CDP Data Analyst study resources

Broadcast Join After a Regional Backfill

Cloudera CDP Data Analyst · Put your knowledge to work in context.

All case studies

Broadcast Join After a Regional Backfill

A retail analytics team runs a nightly Impala query that joins a 4.2-billion-row sales fact table (Parquet, partitioned by sale_date, 1,460 partitions) to a 90,000-row store dimension. Until last week it finished in about 45 seconds. After the team backfilled 90 days of a newly onboarded region, the same query takes over 22 minutes: one executor host shows 100% CPU while the other 15 sit near idle, and the query profile shows 4.1 billion rows broadcast from the fact side. The new region's data also contains roughly 310 million rows where store_id = '-1', a placeholder written by the POS feed. SELECT COUNT(*) on the new partitions is fast, and HDFS is healthy. The team wants to know whether to add memory to the executors, rewrite in Hive, or something else.

1. Before touching any memory settings, what do you check first, and what change do you expect to make? Explain why more executor memory would not fix this.

Read the worked answer

Check statistics and the plan, not memory.

  1. Run EXPLAIN on the query and look for a BROADCAST distribution on the fact side. A 4.1-billion-row broadcast is a planner error, not a capacity error.
  2. Inspect SHOW TABLE STATS sales_fact PARTITION (...) and SHOW COLUMN STATS for the backfilled partitions and for the store dimension. Backfilling with INSERT creates files but does not compute stats; with no stats Impala falls back to default row counts and can decide the fact table is the small side.
  3. Fix: run COMPUTE INCREMENTAL STATS sales_fact PARTITION (...) for the 90 new partitions (or COMPUTE STATS sales_fact if incremental stats are not enabled for that table), plus COMPUTE STATS store_dim if it changed. Then re-run EXPLAIN and confirm the broadcast is gone.

Why memory is the wrong lever: the cost here is distributing a huge build side to every node and then probing it, which multiplies network traffic and per-node work regardless of heap size. More memory lets a bad plan finish, slower and more expensively. The only setting worth revisiting is whether a broadcast hint was hard-coded in the view SQL - if so, remove it. After the plan is correct, runtime filters should also start helping the probe side.

2. After stats are refreshed the broadcast disappears, but the query still takes 6 minutes and one fragment processes about 310 million rows. What is happening, and what do you do? Would a join hint fix it?

Read the worked answer

This is key skew, a different problem from stale stats. In a partitioned hash join all rows with store_id = '-1' hash to the same partition, so one fragment receives 310M rows while the rest receive thousands. The plan is now correct; the data distribution is not.

Options, in order of preference:

  1. Handle the placeholder upstream. If '-1' means 'unknown store', insert a real dimension row with store_id = -1 and keep the rows, or filter them out before the join if the business agrees they are unattributable. Filtering is cheapest; adding the dimension row preserves revenue totals.
  2. Aggregate before joining. Pre-aggregate fact rows by store_id (and any other grouping) in a subquery so the join moves far fewer rows.
  3. Split the query. Process the '-1' rows in a separate branch and UNION ALL the results, so the main join never sees the hot key.

Hints do not solve this: STRAIGHT_JOIN changes join order and distribution, not key-level skew; a broadcast hint would make things worse. Impala has no automatic skew handling comparable to Spark AQE, so the fix must be in the data or the SQL shape. Also check whether other sentinel values (-2, 0) exist, and add a data-quality check that alerts when a single key exceeds a few percent of a partition.

3. Under what conditions would you move this nightly job to Hive on Tez instead of keeping it in Impala? Give concrete criteria rather than 'when it is slow'.

Read the worked answer

Impala is an MPP engine tuned for interactive latency over well-structured data: it keeps intermediates in memory, spills when it must, and a failure restarts the query. Hive on Tez is a batch engine: it spills freely, retries failed tasks instead of restarting the whole query, and handles long multi-stage plans with many shuffles.

Concrete criteria for moving this job:

  • Runtime and shape: p95 above roughly 10 minutes, or the plan needs two or more full scans of the 4.2B-row table with multiple joins, DISTINCTs, or window functions. Impala's fixed memory limits make these spill-heavy and fragile.
  • Fault tolerance: on a busy shared cluster, a mid-query executor loss in Impala means starting over; for a 30-minute job that is unacceptable.
  • Write pattern: the job must MERGE into an ACID table (MERGE INTO in Hive 3) rather than overwrite partitions.
  • Concurrency trade-off: the job competes with BI users for Impala admission-control slots; batch work belongs in the batch queue.

Keep the final aggregation that feeds dashboards in Impala (or in a Kudu/Parquet summary table refreshed by the Hive job). The usual pattern is Hive for the heavy transform, Impala for the last-mile query - not one engine for both.