nsivabalan commented on code in PR #20152: URL: https://github.com/apache/hudi/pull/20152#discussion_r4284080905
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableLayoutAnalyzer.java: ########## @@ -0,0 +1,1825 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities; + +import org.apache.hudi.avro.model.HoodieMetadataColumnStats; +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.hadoop.fs.HadoopFSUtils; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.metadata.MetadataPartitionType; +import org.apache.hudi.storage.StorageConfiguration; +import org.apache.hudi.storage.StoragePath; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.UniformReservoir; +import lombok.extern.slf4j.Slf4j; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.spark.api.java.JavaSparkContext; + +import javax.annotation.Nullable; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.time.LocalDate; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Analyzes the physical layout of a Hudi table: how bytes and base files are spread across + * partitions, how skewed that spread is, and whether the layout shows the classic symptoms + * that make queries and table services slow. + * + * <p>The tool is strictly read-only. It reads the timeline and the file listing (through the + * metadata table when the table has one) and prints a report; it never writes to the table. + * + * <h3>What it reports</h3> + * <ul> + * <li>Table-level base-file size distribution (count, total, min / max / mean / median, + * p50 / p90 / p95 / p99) and the per-partition file-count distribution.</li> + * <li>Partition-size skew: coefficient of variation, Gini coefficient, largest / top-10 + * partition share, and partitions above mean + 2 sigma.</li> + * <li>With {@code --enable-partition-stats}, one row per partition sorted by total bytes + * (capped with {@code --top-n}).</li> + * <li>With {@code --include-row-counts}, per-partition record counts, taken from the metadata + * table's column stats when present and from Parquet footers otherwise.</li> + * <li>With {@code --analyze-table-characteristics}, three detectors, each with a verdict, + * the thresholds it used, and actionable findings: micro-partitioning, small-file + * pile-up (CLEAN / MODERATE / SEVERE / SKIPPED), and hot partitions by recent write + * volume.</li> + * </ul> + * + * <p>Date filtering ({@code --num-days}, {@code --start-date}, {@code --end-date}) narrows the + * partitions considered and only works on tables whose partition names are of the form + * {@code [column=]yyyy-M-d} or {@code [column=]yyyy/M/d}; the half-open interval + * {@code [start, end)} is used. {@code --start-date} takes precedence over {@code --num-days}. + * + * <h3>Example</h3> + * <pre> + * spark-submit --master "local[2]" \ + * --class org.apache.hudi.utilities.HoodieTableLayoutAnalyzer \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.12-1.3.0-SNAPSHOT.jar \ + * --base-path s3://bucket/path/to/table \ + * --enable-partition-stats --analyze-table-characteristics --output JSON + * </pre> + */ +@Slf4j +public class HoodieTableLayoutAnalyzer implements Serializable { + + private static final long serialVersionUID = 1L; + + // Date formatter for parsing partition dates (example: 2023/5/5/ or 2023-5-5). + private static final DateTimeFormatter DATE_FORMATTER = + (new DateTimeFormatterBuilder()).appendOptional(DateTimeFormatter.ofPattern("yyyy/M/d")).appendOptional(DateTimeFormatter.ofPattern("yyyy-M-d")).toFormatter(); + + // File size stats will be displayed in the units specified below. + private static final String[] FILE_SIZE_UNITS = {"B", "KB", "MB", "GB", "TB"}; + + // The record key column is required and populated exactly once per row, so its value count + // in the metadata table's column stats equals the row count of the file. + private static final String ROW_COUNT_COLUMN = "_hoodie_record_key"; + + static final String MICRO_PARTITION_DETECTOR = "micro-partition"; + static final String SMALL_FILES_DETECTOR = "small-files"; + static final String HOT_PARTITIONS_DETECTOR = "hot-partitions"; + + // Spark context + private transient JavaSparkContext jsc; + // config + private final Config cfg; + // Properties with source, hoodie client, key generator etc. + private final TypedProperties props; + + public HoodieTableLayoutAnalyzer(JavaSparkContext jsc, Config cfg) { + this.jsc = jsc; + this.cfg = cfg; + + this.props = cfg.propsFilePath == null + ? UtilHelpers.buildProperties(cfg.configs) + : readConfigFromFileSystem(jsc, cfg); + } + + /** + * Reads config from the file system. + * + * @param jsc {@link JavaSparkContext} instance. + * @param cfg {@link Config} instance. + * @return the {@link TypedProperties} instance. + */ + private TypedProperties readConfigFromFileSystem(JavaSparkContext jsc, Config cfg) { + return UtilHelpers.readConfig(jsc.hadoopConfiguration(), new Path(cfg.propsFilePath), cfg.configs) + .getProps(true); + } + + public static class Config implements Serializable { + + private static final long serialVersionUID = 1L; + + @Parameter(names = {"--base-path", "-bp"}, description = "Base path for the table", required = false) + public String basePath = null; + + @Parameter(names = {"--num-days", "-nd"}, description = "Consider files modified within this many days.", required = false) + public long numDays = 0; + + @Parameter(names = {"--start-date", "-sd"}, description = "Consider files modified on or after this date.", required = false) + public String startDate = null; + + @Parameter(names = {"--end-date", "-ed"}, description = "Consider files modified before this date.", required = false) + public String endDate = null; + + @Parameter(names = {"--enable-table-stats", "-fs"}, description = "Show file-level stats.", required = false) + public boolean tableStats = false; + + @Parameter(names = {"--enable-partition-stats", "-ps"}, description = "Show partition-level stats.", required = false) + public boolean partitionStats = false; + + @Parameter(names = {"--output", "-o"}, description = "Output format: TABLE (default) or JSON.", required = false) + public String output = "TABLE"; + + @Parameter(names = {"--top-n", "-tn"}, + description = "When --enable-partition-stats is set, only show the top N largest partitions (sorted by totalBytes desc). Default 0 = show all.", + required = false) + public int topN = 0; + + @Parameter(names = {"--include-row-counts", "-irc"}, + description = "Opt-in: also collect per-partition row counts. Uses metadata table column stats when present, " + + "Parquet footer reads otherwise. Adds numRecords + avgRowSize columns.", + required = false) + public boolean includeRowCounts = false; + + @Parameter(names = {"--analyze-table-characteristics", "-atc"}, + description = "Opt-in: run micro-partition / small-file / hot-partition detectors and emit a verdict section.", + required = false) + public boolean analyzeTableCharacteristics = false; + + @Parameter(names = {"--print-top-k"}, + description = "When >0 and --analyze-table-characteristics is set, additionally print the top K " + + "partitions with the smallest avg file size under the micro and small-file detectors. Default 0 = off.", + required = false) + public int printTopK = 0; + + @Parameter(names = {"--micro-partition-count-threshold"}, + description = "Table is flagged micro-partitioned when numPartitions exceeds this. Default 10000.", + required = false) + public int microPartitionCountThreshold = 10000; + + @Parameter(names = {"--micro-partition-min-files"}, + description = "Per-partition file count gate for the size-based micro-partition rule. Default 25.", + required = false) + public int microPartitionMinFiles = 25; + + @Parameter(names = {"--micro-partition-max-avg-bytes"}, + description = "Per-partition avg-file-size threshold (bytes) for the size-based rule. Default 52428800 (50 MB).", + required = false) + public long microPartitionMaxAvgBytes = 50L * 1024 * 1024; + + @Parameter(names = {"--micro-partition-min-age-days"}, + description = "Minimum table age before applying the size-based micro-partition rule. Default 30.", + required = false) + public int microPartitionMinAgeDays = 30; + + @Parameter(names = {"--small-files-min-files-per-partition"}, + description = "Per-partition file count gate for small-files detection; partitions with fewer files are excluded from the prevalence ratio. Default 5.", + required = false) + public int smallFilesMinFilesPerPartition = 5; + + @Parameter(names = {"--small-files-threshold-bytes"}, + description = "Avg-file-size threshold (bytes) for flagging a qualifying partition. Default 52428800 (50 MB).", + required = false) + public long smallFilesThresholdBytes = 50L * 1024 * 1024; + + @Parameter(names = {"--small-files-moderate-pct"}, + description = "Fraction of qualifying partitions flagged to trigger MODERATE verdict. Default 0.10.", + required = false) + public double smallFilesModeratePct = 0.10; + + @Parameter(names = {"--small-files-severe-pct"}, + description = "Fraction of qualifying partitions flagged to trigger SEVERE verdict. Default 0.30.", + required = false) + public double smallFilesSeverePct = 0.30; + + @Parameter(names = {"--small-files-min-table-commits"}, + description = "Minimum total ingest commits (active+archived) before emitting the small-file verdict. Default 10.", + required = false) + public int smallFilesMinTableCommits = 10; + + @Parameter(names = {"--hot-window-commits"}, + description = "Number of recent ingest commits to scan for hot-partition detection. Default 50.", + required = false) + public int hotWindowCommits = 50; + + @Parameter(names = {"--hot-partition-commit-share"}, + description = "Minimum share of the hot window a partition must appear in to be flagged hot. Default 0.5.", + required = false) + public double hotPartitionCommitShare = 0.5; + + @Parameter(names = {"--props-path", "-pp"}, description = "Properties file containing base paths one per line", required = false) + public String propsFilePath = null; + + @Parameter(names = {"--parallelism", "-pl"}, description = "Parallelism for valuation", required = false) + public int parallelism = 200; + + @Parameter(names = {"--spark-master", "-ms"}, description = "Spark master", required = false) + public String sparkMaster = null; + + @Parameter(names = {"--spark-memory", "-sm"}, description = "spark memory to use", required = false) + public String sparkMemory = "1g"; + + @Parameter(names = {"--hoodie-conf"}, description = "Any configuration that can be set in the properties file " + + "(using the CLI parameter \"--props\") can also be passed command line using this parameter. This can be repeated", + splitter = IdentitySplitter.class) + public List<String> configs = new ArrayList<>(); + + @Parameter(names = {"--help", "-h"}, help = true) + public Boolean help = false; + + @Override + public String toString() { + return "HoodieTableLayoutAnalyzer {\n" + + " --base-path " + basePath + ", \n" + + " --num-days " + numDays + ", \n" + + " --start-date " + startDate + ", \n" + + " --end-date " + endDate + ", \n" + + " --enable-table-stats " + tableStats + ", \n" + + " --enable-partition-stats " + partitionStats + ", \n" + + " --output " + output + ", \n" + + " --top-n " + topN + ", \n" + + " --include-row-counts " + includeRowCounts + ", \n" + + " --analyze-table-characteristics " + analyzeTableCharacteristics + ", \n" + + " --print-top-k " + printTopK + ", \n" + + " --micro-partition-count-threshold " + microPartitionCountThreshold + ", \n" + + " --micro-partition-min-files " + microPartitionMinFiles + ", \n" + + " --micro-partition-max-avg-bytes " + microPartitionMaxAvgBytes + ", \n" + + " --micro-partition-min-age-days " + microPartitionMinAgeDays + ", \n" + + " --small-files-min-files-per-partition " + smallFilesMinFilesPerPartition + ", \n" + + " --small-files-threshold-bytes " + smallFilesThresholdBytes + ", \n" + + " --small-files-moderate-pct " + smallFilesModeratePct + ", \n" + + " --small-files-severe-pct " + smallFilesSeverePct + ", \n" + + " --small-files-min-table-commits " + smallFilesMinTableCommits + ", \n" + + " --hot-window-commits " + hotWindowCommits + ", \n" + + " --hot-partition-commit-share " + hotPartitionCommitShare + ", \n" + + " --parallelism " + parallelism + ", \n" + + " --spark-master " + sparkMaster + ", \n" + + " --spark-memory " + sparkMemory + ", \n" + + " --props " + propsFilePath + ", \n" + + " --hoodie-conf " + configs + + "\n}"; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Config config = (Config) o; + return Objects.equals(basePath, config.basePath) + && Objects.equals(numDays, config.numDays) + && Objects.equals(startDate, config.startDate) + && Objects.equals(endDate, config.endDate) + && Objects.equals(tableStats, config.tableStats) + && Objects.equals(partitionStats, config.partitionStats) + && Objects.equals(output, config.output) + && Objects.equals(topN, config.topN) + && Objects.equals(includeRowCounts, config.includeRowCounts) + && Objects.equals(analyzeTableCharacteristics, config.analyzeTableCharacteristics) + && Objects.equals(printTopK, config.printTopK) + && Objects.equals(microPartitionCountThreshold, config.microPartitionCountThreshold) + && Objects.equals(microPartitionMinFiles, config.microPartitionMinFiles) + && Objects.equals(microPartitionMaxAvgBytes, config.microPartitionMaxAvgBytes) + && Objects.equals(microPartitionMinAgeDays, config.microPartitionMinAgeDays) + && Objects.equals(smallFilesMinFilesPerPartition, config.smallFilesMinFilesPerPartition) + && Objects.equals(smallFilesThresholdBytes, config.smallFilesThresholdBytes) + && Objects.equals(smallFilesModeratePct, config.smallFilesModeratePct) + && Objects.equals(smallFilesSeverePct, config.smallFilesSeverePct) + && Objects.equals(smallFilesMinTableCommits, config.smallFilesMinTableCommits) + && Objects.equals(hotWindowCommits, config.hotWindowCommits) + && Objects.equals(hotPartitionCommitShare, config.hotPartitionCommitShare) + && Objects.equals(parallelism, config.parallelism) + && Objects.equals(sparkMaster, config.sparkMaster) + && Objects.equals(sparkMemory, config.sparkMemory) + && Objects.equals(propsFilePath, config.propsFilePath) + && Objects.equals(configs, config.configs); + } + + @Override + public int hashCode() { + return Objects.hash(basePath, numDays, startDate, endDate, tableStats, partitionStats, output, topN, includeRowCounts, + analyzeTableCharacteristics, printTopK, microPartitionCountThreshold, microPartitionMinFiles, microPartitionMaxAvgBytes, + microPartitionMinAgeDays, smallFilesMinFilesPerPartition, smallFilesThresholdBytes, smallFilesModeratePct, + smallFilesSeverePct, smallFilesMinTableCommits, hotWindowCommits, hotPartitionCommitShare, + parallelism, sparkMaster, sparkMemory, propsFilePath, configs, help); + } + } + + public static void main(String[] args) { + final Config cfg = new Config(); + JCommander cmd = new JCommander(cfg, null, args); + + if (cfg.help || args.length == 0) { + cmd.usage(); + System.exit(1); + } + + Map<String, String> sparkConfigMap = new HashMap<>(); + sparkConfigMap.put("spark.executor.memory", cfg.sparkMemory); + JavaSparkContext jsc = UtilHelpers.buildSparkContext("Table-Layout-Analyzer", cfg.sparkMaster, false, sparkConfigMap); + + try { + HoodieTableLayoutAnalyzer analyzer = new HoodieTableLayoutAnalyzer(jsc, cfg); + analyzer.run(); + } catch (TableNotFoundException e) { + log.warn("The Hudi data table is not found: [{}].", cfg.basePath, e); + } catch (Throwable throwable) { + log.error("Failed to analyze table layout for {}", cfg, throwable); + } finally { + jsc.stop(); + } + } + + public void run() { + try { + log.info(cfg.toString()); + log.info(" ****** Analyzing table layout ******"); + + // Determine starting and ending date intervals for filtering data files. + LocalDate[] dateInterval = getUserSpecifiedDateInterval(cfg); + + if (cfg.propsFilePath != null) { + List<String> filePaths = getFilePaths(cfg.propsFilePath, jsc.hadoopConfiguration()); + for (String filePath : filePaths) { + analyzeTable(filePath, dateInterval); + } + } else { + if (cfg.basePath == null) { + throw new HoodieIOException("Base path needs to be set."); + } + analyzeTable(cfg.basePath, dateInterval); + } + + } catch (Exception e) { + throw new HoodieException("Unable to analyze table layout for " + cfg.basePath, e); + } + } + + private void analyzeTable(String basePath, LocalDate[] dateInterval) throws IOException { + log.info("Processing table {}", basePath); + StorageConfiguration<?> storageConf = HadoopFSUtils.getStorageConfWithCopy(jsc.hadoopConfiguration()); + HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder() + .setBasePath(basePath) + .setConf(storageConf.newInstance()).build(); + HoodieTableConfig tableConfig = metaClient.getTableConfig(); + boolean mdtEnabled = tableConfig.isMetadataPartitionAvailable(MetadataPartitionType.FILES); + boolean colStatsAvailable = tableConfig.isMetadataPartitionAvailable(MetadataPartitionType.COLUMN_STATS); + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder() + .enable(mdtEnabled) + .build(); + HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc); + + // Both tableMetadata and fileSystemView are AutoCloseable. Under --props-path batch mode + // (multiple tables per process) leaving them open leaks metadata readers and file-listing + // handles per table, which can hit FD / heap limits over long batches. + // + // The file system view is built once per table and honors the table's metadata-table + // setting, so on large tables partitions are listed from the metadata table instead of + // one storage listing per partition. + try (HoodieTableMetadata tableMetadata = metaClient.getTableFormat().getMetadataFactory() + .create(engineContext, metaClient.getStorage(), metadataConfig, basePath); + HoodieTableFileSystemView fileSystemView = FileSystemViewManager + .createInMemoryFileSystemView(new HoodieLocalEngineContext(storageConf), metaClient, metadataConfig)) { + + List<String> allPartitions = tableMetadata.getAllPartitionPaths(); + + // As a sanity check, throw exception and exit early if date interval is specified, but the first partition does not have + // date. + if (dateInterval != null && !allPartitions.isEmpty() && getPartitionDate(allPartitions.get(0)) == null) { + throw new HoodieException( + "Cannot apply --start-date, --end-date, or --num-days when partition does not contain date. Interval: " + Arrays.toString(dateInterval) + ", Partition Name: " + allPartitions.get(0)); + } + + List<PartitionRow> rows = new ArrayList<>(); + // Table-level reservoirs are one per run, so 1M slots (~8 MB each) is fine and gives + // accurate p95/p99 on tables with hundreds of thousands of files. + final Histogram tableSizeHistogram = new Histogram(new UniformReservoir(1_000_000)); + final Histogram tableFileCountHistogram = new Histogram(new UniformReservoir(1_000_000)); + for (String partition : allPartitions) { + if (!isPartitionInRange(partition, dateInterval)) { + continue; + } + PartitionRow row = collectPartitionRow(partition, fileSystemView, tableMetadata, colStatsAvailable, storageConf); + row.sizes.forEach(tableSizeHistogram::update); + tableFileCountHistogram.update(row.fileCount); + rows.add(row); + } + + // Sort partitions by total bytes descending so output (and --top-n) shows the largest first. + rows.sort(Comparator.comparingLong((PartitionRow r) -> r.totalBytes).reversed()); + + TableCharacteristics characteristics = cfg.analyzeTableCharacteristics + ? analyzeCharacteristics(metaClient, rows) + : null; + + OutputFormat output = parseOutputFormat(cfg.output); + if (output == OutputFormat.JSON) { + emitJson(basePath, rows, tableSizeHistogram, tableFileCountHistogram, mdtEnabled, characteristics); + } else { + emitTable(basePath, rows, tableSizeHistogram, tableFileCountHistogram, mdtEnabled, characteristics); + } + } catch (HoodieException he) { + throw he; + } catch (Exception e) { + throw new HoodieException("Failure while analyzing table layout for " + basePath, e); + } + } + + /** + * Applies the optional date interval to a partition name. A partition is in range when: + * 1. no interval is specified, or the partition name does not contain a date; + * 2. only a start date is specified and the partition date is on or after it; + * 3. only an end date is specified and the partition date is before it; + * 4. both are specified and the partition date lies in [startDate, endDate). + */ + private static boolean isPartitionInRange(String partition, LocalDate[] dateInterval) { + if (dateInterval == null) { + return true; + } + LocalDate partitionDate = getPartitionDate(partition); + if (partitionDate == null) { + return true; + } + LocalDate startDate = dateInterval[0]; + LocalDate endDate = dateInterval[1]; + boolean afterStart = startDate == null || !partitionDate.isBefore(startDate); + boolean beforeEnd = endDate == null || partitionDate.isBefore(endDate); + return afterStart && beforeEnd; + } + + private PartitionRow collectPartitionRow(String partition, HoodieTableFileSystemView fileSystemView, + HoodieTableMetadata tableMetadata, boolean colStatsAvailable, + StorageConfiguration<?> storageConf) { + List<HoodieBaseFile> baseFiles = fileSystemView.getLatestBaseFiles(partition).collect(Collectors.toList()); + + // The per-partition histogram is only rendered under --enable-partition-stats. Skip the + // allocation when it isn't needed: UniformReservoir eagerly allocates an AtomicLongArray + // of its capacity (8 bytes per slot) on construction, so a 1M-slot reservoir is ~8 MB + // per partition and would explode driver heap on tables with many partitions. When we + // do build it, use a 4096-slot reservoir, plenty of headroom for the per-partition + // file count in practice, while keeping the per-partition cost at ~32 KB. + Histogram partitionSizeHist = cfg.partitionStats ? new Histogram(new UniformReservoir(4096)) : null; + long partitionTotalBytes = 0L; + List<Long> sizes = new ArrayList<>(baseFiles.size()); + for (HoodieBaseFile baseFile : baseFiles) { + long size = baseFile.getFileSize(); + if (partitionSizeHist != null) { + partitionSizeHist.update(size); + } + sizes.add(size); + partitionTotalBytes += size; + } + + Long partitionRowCount = null; + if (cfg.includeRowCounts && !baseFiles.isEmpty()) { + partitionRowCount = computeRowCount(partition, baseFiles, tableMetadata, colStatsAvailable, storageConf); + } + return new PartitionRow(partition, baseFiles.size(), partitionTotalBytes, sizes, partitionSizeHist, partitionRowCount); + } + + // ---- table characteristics (micro / small / hot detectors) ---------------- + + /** + * Runs the three table-characteristic detectors over the per-partition row set. Each + * detector's verdict is independent; the umbrella flag {@code --analyze-table-characteristics} + * gates whether any of them run, and thresholds are individually configurable. + */ + private TableCharacteristics analyzeCharacteristics(HoodieTableMetaClient metaClient, List<PartitionRow> rows) { + TableCharacteristics tc = new TableCharacteristics(); + tc.tableType = metaClient.getTableType(); + tc.tableAgeDays = computeTableAgeDays(metaClient); + tc.numPartitions = rows.size(); + + runMicroPartitionDetector(rows, tc); + runSmallFileDetector(metaClient, rows, tc); + + try { + computeHotPartitions(metaClient, tc); + } catch (Exception e) { + log.warn("Hot-partition detection failed: {}", e.getMessage()); + tc.hotPartitionDetectionFailed = true; + } + return tc; + } + + /** + * Micro-partition verdict: table is "micro" when partition count exceeds the count threshold, + * OR (on mature tables) any partition has many small files. The size rule is gated by + * table age, since new tables legitimately have small partitions. + */ + private void runMicroPartitionDetector(List<PartitionRow> rows, TableCharacteristics tc) { + boolean countTrigger = tc.numPartitions > cfg.microPartitionCountThreshold; + boolean sizeTriggerEligible = tc.tableAgeDays >= cfg.microPartitionMinAgeDays; + int sizeMatches = 0; + if (sizeTriggerEligible) { + for (PartitionRow r : rows) { + if (r.fileCount >= cfg.microPartitionMinFiles && r.avgFileSize() < cfg.microPartitionMaxAvgBytes) { + sizeMatches++; + } + } + } + tc.microCountTrigger = countTrigger; + tc.microSizeTrigger = sizeMatches > 0; + tc.microSizeMatchCount = sizeMatches; + tc.microPartitioned = countTrigger || tc.microSizeTrigger; + // Top-K evidence: partitions with the smallest avg file size among those satisfying the + // micro file-count gate. Independent of the verdict; useful even when the verdict is "no". + if (cfg.printTopK > 0) { + tc.microTopKSmallest.addAll(collectTopKSmallest(rows, cfg.microPartitionMinFiles, cfg.printTopK)); + } + } + + /** + * Small-file pile-up verdict (CLEAN / MODERATE / SEVERE / SKIPPED) based on per-partition + * prevalence. Count partitions with >= minFiles as "qualifying"; among those, count partitions + * with avgFileSize < threshold as "flagged". Verdict thresholds are flagged/qualifying ratio + * against the moderate/severe pct gates. Skipped when the table has too few commits, since + * the signal is unreliable on freshly-ingested tables. + */ + private void runSmallFileDetector(HoodieTableMetaClient metaClient, List<PartitionRow> rows, TableCharacteristics tc) { + long totalIngestCommits = countTotalIngestCommits(metaClient); + tc.smallFileTableIngestCommitCount = totalIngestCommits; + if (totalIngestCommits < cfg.smallFilesMinTableCommits) { + tc.smallFileVerdict = SmallFileVerdict.SKIPPED; + return; + } + int qualifying = 0; + int flagged = 0; + for (PartitionRow r : rows) { + if (r.fileCount < cfg.smallFilesMinFilesPerPartition) { + continue; + } + qualifying++; + if (r.avgFileSize() < cfg.smallFilesThresholdBytes) { + flagged++; + } + } + tc.smallFileQualifyingPartitions = qualifying; + tc.smallFileFlaggedPartitions = flagged; + tc.smallFileFlaggedPct = qualifying == 0 ? 0.0 : (double) flagged / qualifying; + tc.smallFileVerdict = classifySmallFileVerdict(qualifying, tc.smallFileFlaggedPct); + // Top-K evidence: partitions with the smallest avg file size among the qualifying set. + if (cfg.printTopK > 0) { + tc.smallFileTopKSmallest.addAll(collectTopKSmallest(rows, cfg.smallFilesMinFilesPerPartition, cfg.printTopK)); + } + } + + /** + * Returns the top-K partitions with the smallest avg file size among those satisfying the + * given file-count gate, in ascending avg-file-size order. + */ + private static List<SmallestPartition> collectTopKSmallest(List<PartitionRow> rows, int minFiles, int k) { + List<SmallestPartition> qualifying = new ArrayList<>(); + for (PartitionRow r : rows) { + if (r.fileCount < minFiles) { + continue; + } + qualifying.add(new SmallestPartition(r.partition, r.fileCount, r.totalBytes, r.avgFileSize())); + } + qualifying.sort(Comparator.comparingLong((SmallestPartition p) -> p.avgFileSize) + .thenComparingLong(p -> p.totalBytes)); + return qualifying.subList(0, Math.min(k, qualifying.size())); + } + + private SmallFileVerdict classifySmallFileVerdict(int qualifying, double flaggedPct) { + if (qualifying == 0) { + return SmallFileVerdict.CLEAN; + } else if (flaggedPct >= cfg.smallFilesSeverePct) { + return SmallFileVerdict.SEVERE; + } else if (flaggedPct >= cfg.smallFilesModeratePct) { + return SmallFileVerdict.MODERATE; + } + return SmallFileVerdict.CLEAN; + } + + /** + * Estimates table age in days as the earlier of two cheap signals: the modification time of + * {@code hoodie.properties} and the first instant on the active timeline. Neither is exact + * on its own. The properties file is rewritten on upgrades and when metadata partitions + * change, so its mtime can be later than table creation; the active timeline's first + * instant moves forward as commits are archived. Taking the earlier of the two never + * over-estimates age, and is only an under-estimate when both signals are later than the + * true creation time. The age gates the size-based micro-partition rule, so an + * under-estimate errs on the side of not flagging. + */ + private long computeTableAgeDays(HoodieTableMetaClient metaClient) { + LocalDate today = LocalDate.now(); + long ageDays = 0; + try { + StoragePath propsPath = new StoragePath(metaClient.getMetaPath(), HoodieTableConfig.HOODIE_PROPERTIES_FILE); + long mtime = metaClient.getStorage().getPathInfo(propsPath).getModificationTime(); + if (mtime > 0) { + LocalDate modified = Instant.ofEpochMilli(mtime).atZone(ZoneId.systemDefault()).toLocalDate(); + ageDays = Math.max(ageDays, ChronoUnit.DAYS.between(modified, today)); + } + } catch (Exception e) { + log.warn("Failed to read hoodie.properties modification time for table age: {}", e.getMessage()); + } + + try { + Option<HoodieInstant> firstOpt = metaClient.getActiveTimeline().firstInstant(); + if (firstOpt.isPresent()) { + String ts = firstOpt.get().requestedTime(); + // Hudi instant times are yyyyMMddHHmmssSSS. Parse the date portion. + if (ts.length() >= 8) { + LocalDate first = LocalDate.parse(ts.substring(0, 8), DateTimeFormatter.ofPattern("yyyyMMdd")); + ageDays = Math.max(ageDays, ChronoUnit.DAYS.between(first, today)); + } + } + } catch (Exception e) { + log.warn("Failed to derive table age from the active timeline: {}", e.getMessage()); + } + return ageDays; + } + + /** + * Counts total completed ingest commits (commit + deltacommit + replacecommit) across + * active and archived timelines. Used to gate the small-file verdict: tables with too + * few commits aren't worth scoring because the signal is dominated by initial-write + * noise. The archived timeline is only consulted when the active timeline alone is under + * the gate, to avoid the archive scan on mature tables. + * + * <p>Replacecommits from clustering are excluded, since they aren't ingests and would + * inflate the count on tables with frequent clustering. INSERT_OVERWRITE / + * INSERT_OVERWRITE_TABLE replacecommits are real ingests and are kept. Distinguishing the + * two requires reading each replacecommit's metadata, which is only done for the active + * timeline: the archived timeline is loaded without instant details, so its replacecommits + * are skipped and the count is a lower bound there. + */ + private long countTotalIngestCommits(HoodieTableMetaClient metaClient) { + try { + long activeCount = countIngestCommits(metaClient.getActiveTimeline(), true); + if (activeCount >= cfg.smallFilesMinTableCommits) { + return activeCount; + } + // Need archived to see if the table has actually received enough commits. + long archivedCount = countIngestCommits(metaClient.getArchivedTimeline(), false); + return activeCount + archivedCount; + } catch (Exception e) { + log.warn("Failed to count ingest commits: {}", e.getMessage()); + return 0; + } + } + + /** + * Counts completed ingest commits on a single timeline. Commit/deltacommit are counted + * unconditionally; each replacecommit is read to distinguish real ingests + * (INSERT_OVERWRITE / INSERT_OVERWRITE_TABLE) from clustering, which is skipped. When + * {@code detailsReadable} is false, replacecommits are skipped rather than read. + */ + private long countIngestCommits(HoodieTimeline timeline, boolean detailsReadable) { + return timeline.filterCompletedInstants().getInstantsAsStream() + .filter(i -> { + String action = i.getAction(); + if (action.equals(HoodieTimeline.COMMIT_ACTION) || action.equals(HoodieTimeline.DELTA_COMMIT_ACTION)) { Review Comment: Correct on both, fixed in `1ea199b3d6b4`, and `WriteOperationType.isTableService(op)` was the right answer for each. The javadoc did say table-service commits should be excluded, but the filter returned `true` for any `COMMIT_ACTION` — and on Merge-on-Read a completed compaction is written as a plain commit, so it was counted while clustering was correctly skipped. Inconsistent in exactly the way you describe. Both are now excluded by operation type. The hot-partition window had the same gap for a different reason: naming `COMPACT` and `CLUSTER` individually missed `LOG_COMPACT`, which inflated the window on log-compacted MOR tables. `isTableService` covers that and `INDEX` too. One behaviour I kept: when commit bodies are unreadable the operation type is unknowable, so commits and deltacommits are still counted and replacecommits skipped. Undercounting beats misclassifying, and the javadoc now says so. Test added: three UPSERTs plus one COMPACT, with the ingest gate set to 4, asserts the count stays at 3. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableLayoutAnalyzer.java: ########## @@ -0,0 +1,1825 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities; + +import org.apache.hudi.avro.model.HoodieMetadataColumnStats; +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.hadoop.fs.HadoopFSUtils; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.metadata.MetadataPartitionType; +import org.apache.hudi.storage.StorageConfiguration; +import org.apache.hudi.storage.StoragePath; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.UniformReservoir; +import lombok.extern.slf4j.Slf4j; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.spark.api.java.JavaSparkContext; + +import javax.annotation.Nullable; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.time.LocalDate; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Analyzes the physical layout of a Hudi table: how bytes and base files are spread across + * partitions, how skewed that spread is, and whether the layout shows the classic symptoms + * that make queries and table services slow. + * + * <p>The tool is strictly read-only. It reads the timeline and the file listing (through the + * metadata table when the table has one) and prints a report; it never writes to the table. + * + * <h3>What it reports</h3> + * <ul> + * <li>Table-level base-file size distribution (count, total, min / max / mean / median, + * p50 / p90 / p95 / p99) and the per-partition file-count distribution.</li> + * <li>Partition-size skew: coefficient of variation, Gini coefficient, largest / top-10 + * partition share, and partitions above mean + 2 sigma.</li> + * <li>With {@code --enable-partition-stats}, one row per partition sorted by total bytes + * (capped with {@code --top-n}).</li> + * <li>With {@code --include-row-counts}, per-partition record counts, taken from the metadata + * table's column stats when present and from Parquet footers otherwise.</li> + * <li>With {@code --analyze-table-characteristics}, three detectors, each with a verdict, + * the thresholds it used, and actionable findings: micro-partitioning, small-file + * pile-up (CLEAN / MODERATE / SEVERE / SKIPPED), and hot partitions by recent write + * volume.</li> + * </ul> + * + * <p>Date filtering ({@code --num-days}, {@code --start-date}, {@code --end-date}) narrows the + * partitions considered and only works on tables whose partition names are of the form + * {@code [column=]yyyy-M-d} or {@code [column=]yyyy/M/d}; the half-open interval + * {@code [start, end)} is used. {@code --start-date} takes precedence over {@code --num-days}. + * + * <h3>Example</h3> + * <pre> + * spark-submit --master "local[2]" \ + * --class org.apache.hudi.utilities.HoodieTableLayoutAnalyzer \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.12-1.3.0-SNAPSHOT.jar \ + * --base-path s3://bucket/path/to/table \ + * --enable-partition-stats --analyze-table-characteristics --output JSON + * </pre> + */ +@Slf4j +public class HoodieTableLayoutAnalyzer implements Serializable { + + private static final long serialVersionUID = 1L; + + // Date formatter for parsing partition dates (example: 2023/5/5/ or 2023-5-5). + private static final DateTimeFormatter DATE_FORMATTER = + (new DateTimeFormatterBuilder()).appendOptional(DateTimeFormatter.ofPattern("yyyy/M/d")).appendOptional(DateTimeFormatter.ofPattern("yyyy-M-d")).toFormatter(); + + // File size stats will be displayed in the units specified below. + private static final String[] FILE_SIZE_UNITS = {"B", "KB", "MB", "GB", "TB"}; + + // The record key column is required and populated exactly once per row, so its value count + // in the metadata table's column stats equals the row count of the file. + private static final String ROW_COUNT_COLUMN = "_hoodie_record_key"; + + static final String MICRO_PARTITION_DETECTOR = "micro-partition"; + static final String SMALL_FILES_DETECTOR = "small-files"; + static final String HOT_PARTITIONS_DETECTOR = "hot-partitions"; + + // Spark context + private transient JavaSparkContext jsc; + // config + private final Config cfg; + // Properties with source, hoodie client, key generator etc. + private final TypedProperties props; + + public HoodieTableLayoutAnalyzer(JavaSparkContext jsc, Config cfg) { + this.jsc = jsc; + this.cfg = cfg; + + this.props = cfg.propsFilePath == null Review Comment: Confirmed and removed in `1ea199b3d6b4`. `props` was assigned in the constructor and never read again, so `--hoodie-conf` was accepted, parsed and discarded. Its own help text referred to a `--props` option this tool does not have, and `toString()` printed `--props-path` under that name too. I removed the flag and the field rather than wiring it up: the analyzer only walks the timeline and lists files, building its own metaClient from `--base-path`, so there is no consumer for arbitrary `hoodie.*` properties. Inventing one to justify the flag seemed worse than deleting it. A user passing `--hoodie-conf` now gets an "unknown option" error instead of silence. The tool is new in this PR, so nothing can depend on it. On the overloading: `--props-path` is left as-is, serving the base-path list its `@Parameter` description documents. The properties-file reading at `:157` was the other half of the dead code path and went with it, so the flag now has exactly one meaning. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
