Three Thousand Small CSV Files and a Duplicate Feed

No. hive.merge.mapfiles, hive.merge.smallfiles.avgsize, and hive.merge.mapredfiles are write-time options for Hive jobs: they merge the files that a Hive query or INSERT produces. The application writes directly to HDFS, so Hive never sees those files as output and will never rewrite them. Re-running the dashboard query changes nothing on disk.

Even a compaction job that copied the files would leave the bigger problem: OpenCSVSerde means text parsing of every column, no column pruning, and no predicate pushdown on non-partition columns. The 18-minute query is paying for reading all columns of all ~12 GB per day of text.

The real fix is a one-time rewrite into a columnar format, plus control of file size:

1. Create a new external Parquet table with the same partition spec (dt, and hour if hour-level pruning is actually used).
2. Rewrite with a deduplicating INSERT that controls output file count - DISTRIBUTE BY dt, hour (or a repartition to roughly 128–256 MB per file) so each reducer writes one or a few files instead of thousands.
3. Point a view at the new table, validate, and keep the old table for a rollback window.

With Parquet and ~200 MB files, a day is a few dozen files instead of 3,100, and the GROUP BY reads only event_type plus the partition column.

INSERT OVERWRITE TABLE events_clean PARTITION (dt = '${dt}')
SELECT event_id, user_id, event_type, event_ts, ingestion_ts, ...
FROM (
  SELECT e.*,
         ROW_NUMBER() OVER (
           PARTITION BY event_id
           ORDER BY ingestion_ts DESC, hour DESC
         ) AS rn
  FROM events_staging e
  WHERE e.dt = '${dt}'
  DISTRIBUTE BY event_id
) t
WHERE rn = 1;

Why DISTRIBUTE BY: a window function requires all rows of a partition to be in one reducer. DISTRIBUTE BY event_id guarantees that; without it Hive may spread the same event_id across reducers and produce two surviving rows. The ORDER BY ingestion_ts DESC keeps the latest retry; add a deterministic tiebreaker (hour) so results are reproducible when timestamps tie.

Why reruns are safe: INSERT OVERWRITE ... PARTITION replaces the partition's contents, so re-running the same ${dt} yields the same result rather than appending duplicates. That idempotency is what makes a backfill safe.

Edge case to flag: a retry can land under a different dt (for example, a batch from 23:50 delivered at 00:10). Deduplicating one day in isolation will miss those. Either widen the window to dt IN (${dt}, date_sub(${dt},1)) and write only matching rows, or run a periodic anti-join against the target table. Validate with COUNT(*) versus COUNT(DISTINCT event_id) on a sample day; the gap should be about 2.5%.

Roll out behind a stable name.

1. Build events_clean alongside the existing table and backfill 90 days.
2. Reconcile before switching: per-day COUNT(*), COUNT(DISTINCT event_id), SUM(revenue), min/max event_ts, and a row-level checksum on a sample day. A 2.5% row reduction from dedup is expected; anything else is a bug.
3. Create a view named events pointing at events_clean, so dashboards keep their existing SQL. If the dashboard cannot be repointed, swap the partition locations under a maintenance window and keep the old directory for rollback.
4. Run old and new in parallel for 3–5 days and diff the daily aggregates.

Ongoing monitoring and maintenance:

• File health per partition: file count and average file size; alert if a day exceeds about 30 files or average size falls below 64 MB.
• Partition registration: run MSCK REPAIR TABLE or an ALTER TABLE ... ADD PARTITION step after every load, or new data will silently be invisible.
• Duplicate ratio: a daily check that dedup removes 2.0–3.0% of rows; a sudden change means the upstream retry behavior changed.
• Query p95 for the top dashboard queries, and NameNode RPC queue length, as the proof the fix worked.
• Layout drift: if the app later adds columns, the SerDe/Parquet schema must be updated or the new fields vanish silently.