nsivabalan commented on code in PR #8645: URL: https://github.com/apache/hudi/pull/8645#discussion_r1188012377
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java: ########## @@ -0,0 +1,411 @@ +/* + * 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.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.SerializableConfiguration; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.metadata.HoodieTableMetadata; + +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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaSparkContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Calculate and output file size stats of data files that were modified in the half-open interval [start date (--start-date parameter), + * end date (--end-date parameter)). --num-days parameter can be used to select data files over last --num-days. If --start-date is + * specified, --num-days will be ignored. If none of the date parameters are set, stats will be computed over all data files of all + * partitions in the table. Note that date filtering is carried out only if the partition name has the format '[column name=]yyyy-M-d', + * '[column name=]yyyy/M/d'. By default, only table level file size stats are printed. If --partition-status option is used, partition + * level file size stats also get printed. + * <br><br> + * The following stats are calculated: + * Number of files. + * Total table size. + * Minimum file size + * Maximum file size + * Average file size + * Median file size + * p50 file size + * p90 file size + * p95 file size + * p99 file size + * <br><br> + * Sample spark-submit command: + * ./bin/spark-submit \ + * --class org.apache.hudi.utilities.TableSizeStats \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.11-0.14.0-SNAPSHOT.jar \ + * --base-path <base-path> \ + * --num-days <number-of-days> + */ +public class TableSizeStats implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TableSizeStats.class); + + // 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"}; + + // Spark context + private transient JavaSparkContext jsc; + // config + private Config cfg; + // Properties with source, hoodie client, key generator etc. + private TypedProperties props; + + public TableSizeStats(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 { + @Parameter(names = {"--base-path", "-sp"}, 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 = {"--partition-stats", "-ps"}, description = "Show partition-level stats besides table-level stats.", required = false) + public boolean partitionStats = false; + + @Parameter(names = {"--props-path", "-pp"}, description = "Properties file containing base paths one per line", required = false) Review Comment: can this work for N tables? ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java: ########## @@ -0,0 +1,411 @@ +/* + * 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.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.SerializableConfiguration; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.metadata.HoodieTableMetadata; + +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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaSparkContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Calculate and output file size stats of data files that were modified in the half-open interval [start date (--start-date parameter), + * end date (--end-date parameter)). --num-days parameter can be used to select data files over last --num-days. If --start-date is + * specified, --num-days will be ignored. If none of the date parameters are set, stats will be computed over all data files of all + * partitions in the table. Note that date filtering is carried out only if the partition name has the format '[column name=]yyyy-M-d', + * '[column name=]yyyy/M/d'. By default, only table level file size stats are printed. If --partition-status option is used, partition + * level file size stats also get printed. + * <br><br> + * The following stats are calculated: + * Number of files. + * Total table size. + * Minimum file size + * Maximum file size + * Average file size + * Median file size + * p50 file size + * p90 file size + * p95 file size + * p99 file size + * <br><br> + * Sample spark-submit command: + * ./bin/spark-submit \ + * --class org.apache.hudi.utilities.TableSizeStats \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.11-0.14.0-SNAPSHOT.jar \ + * --base-path <base-path> \ + * --num-days <number-of-days> + */ +public class TableSizeStats implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TableSizeStats.class); + + // 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"}; + + // Spark context + private transient JavaSparkContext jsc; + // config + private Config cfg; + // Properties with source, hoodie client, key generator etc. + private TypedProperties props; + + public TableSizeStats(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 { + @Parameter(names = {"--base-path", "-sp"}, 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 = {"--partition-stats", "-ps"}, description = "Show partition-level stats besides table-level stats.", required = false) + public boolean partitionStats = false; + + @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 "TableSizeStats {\n" + + " --base-path " + basePath + ", \n" + + " --num-days " + numDays + ", \n" + + " --start-date " + startDate + ", \n" + + " --end-date " + endDate + ", \n" + + " --partition-stats " + partitionStats + ", \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 basePath.equals(config.basePath) + && Objects.equals(numDays, config.numDays) + && Objects.equals(startDate, config.startDate) + && Objects.equals(endDate, config.endDate) + && Objects.equals(partitionStats, config.partitionStats) + && 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, partitionStats, 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); + } + + SparkConf sparkConf = UtilHelpers.buildSparkConf("Table-Size-Stats", cfg.sparkMaster); + sparkConf.set("spark.executor.memory", cfg.sparkMemory); + JavaSparkContext jsc = new JavaSparkContext(sparkConf); + + try { + TableSizeStats tableSizeStats = new TableSizeStats(jsc, cfg); + tableSizeStats.run(); + } catch (TableNotFoundException e) { + LOG.warn(String.format("The Hudi data table is not found: [%s].", cfg.basePath), e); + } catch (Throwable throwable) { + LOG.error("Failed to get table size stats for " + cfg, throwable); + } finally { + jsc.stop(); + } + } + + public void run() { + try { + LOG.info(cfg.toString()); + LOG.info(" ****** Fetching table size stats ******"); + // Set endDate to null by default. + LocalDate endDate = null; + if (cfg.endDate != null) { + try { + endDate = LocalDate.parse(cfg.endDate, DATE_FORMATTER); + LOG.info("Setting ending date to {}. ", endDate); + } catch (DateTimeParseException dtpe) { + throw new HoodieException("Unable to parse --end-date. ", dtpe); + } + } else { + LOG.info("End date is not specified: {}.", endDate); + } + + // Set startDate to null by default. + LocalDate startDate = null; + + // Set startDate to cfg.startDate if specified. cfg.startDate takes priority over cfg.numDays if both are specified. + if (cfg.startDate != null) { + startDate = LocalDate.parse(cfg.startDate, DATE_FORMATTER); + LOG.info("Setting starting date to {}.", startDate); + } else { + if (cfg.numDays == 0) { + LOG.info("Start date not specified: {}.", startDate); + } else if (cfg.numDays > 0) { + endDate = LocalDate.now(); + startDate = endDate.minusDays(cfg.numDays); + LOG.info("Setting starting date to {} ({} - {} days). ", startDate, endDate, cfg.numDays); + } else { + throw new HoodieException("--num-days must specify a positive value."); + } + } + + // Check if starting date is before ending date. + if (startDate != null && endDate != null && !startDate.isBefore(endDate)) { + throw new HoodieException("Starting date must be before ending date. Start Date: " + startDate + ", End Date: " + endDate); + } + + if (cfg.propsFilePath != null) { + List<String> filePaths = getFilePaths(cfg.propsFilePath, jsc.hadoopConfiguration()); + for (String filePath : filePaths) { + logTableStats(filePath, startDate, endDate, cfg.partitionStats); + } + } else { + if (cfg.basePath == null) { + throw new HoodieIOException("Base path needs to be set."); + } + logTableStats(cfg.basePath, startDate, endDate, cfg.partitionStats); + } + + } catch (Exception e) { + throw new HoodieException("Unable to do fetch table size stats." + cfg.basePath, e); + } + } + + private List<String> getFilePaths(String propsPath, Configuration hadoopConf) { + List<String> filePaths = new ArrayList<>(); + FileSystem fs = FSUtils.getFs( + propsPath, + Option.ofNullable(hadoopConf).orElseGet(Configuration::new) + ); + + try (BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(new Path(propsPath))))) { + String line = reader.readLine(); + while (line != null) { + filePaths.add(line); + line = reader.readLine(); + } + } catch (IOException ioe) { + LOG.error("Error reading in properties from dfs from file." + propsPath); + throw new HoodieIOException("Cannot read properties from dfs from file " + propsPath, ioe); + } + return filePaths; + } + + private void logTableStats(String basePath, LocalDate startDate, LocalDate endDate, boolean partitionStats) throws IOException { + + LOG.warn("Processing table " + basePath); + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder() + .enable(false) + .build(); + HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc); + HoodieTableMetadata tableMetadata = HoodieTableMetadata.create(engineContext, metadataConfig, basePath, + FileSystemViewStorageConfig.SPILLABLE_DIR.defaultValue()); + SerializableConfiguration serializableConfiguration = new SerializableConfiguration(jsc.hadoopConfiguration()); + + List<String> allPartitions = tableMetadata.getAllPartitionPaths(); + + final Histogram tableHistogram = new Histogram(new UniformReservoir(1_000_000)); + allPartitions.forEach(partition -> { + // Partition name should conform to date format if startDate and/or endDate are specified. Otherwise, we don't + // need to parse partition name as date. + String dateString = partition; + if (partition.contains("=")) { + // Partition name may be in the form of "<column>=<date>". Try parsing out date. + String[] parts = partition.split("="); + dateString = parts[1].trim(); + } + + LocalDate partitionDate = null; + try { + if (startDate != null || endDate != null) { + partitionDate = LocalDate.parse(dateString, DATE_FORMATTER); + } + LOG.info("Parsed " + partition); + } catch (DateTimeParseException dtpe) { + LOG.error("Partition name {} must conform to date format if --start-date, --end-date, or --num-days are specified. ", partition, dtpe); + } + + // Compute file size stats for all files in this partition if: + // 1. partition date is null (startDate and endDate are both null and not specified) + // 2. endDate is null (not specified) and partition date is equal to or after startDate. + // 3. startDate is null (not specified) and partition date is before endDate. + // 4. startDate and endDate are both specified and partition date lies in the range [startDate, endDate) + if (partitionDate == null + || (endDate == null && (partitionDate.isEqual(startDate) || partitionDate.isAfter(startDate))) + || (startDate == null && (partitionDate.isBefore(endDate))) + || (startDate != null && endDate != null && (partitionDate.isEqual(startDate) || partitionDate.isAfter(startDate) && partitionDate.isBefore(endDate)))) { + HoodieTableMetaClient metaClientLocal = HoodieTableMetaClient.builder() + .setBasePath(basePath) + .setConf(serializableConfiguration.get()).build(); + HoodieMetadataConfig metadataConfig1 = HoodieMetadataConfig.newBuilder() + .enable(false) Review Comment: we should fetch this from table props and use it. ``` boolean metadataEnabled = !this.getTableConfig().getMetadataPartitions().isEmpty() && this.getTableConfig().getMetadataPartitions().contains("files") ``` ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java: ########## @@ -0,0 +1,411 @@ +/* + * 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.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.SerializableConfiguration; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.metadata.HoodieTableMetadata; + +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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaSparkContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Calculate and output file size stats of data files that were modified in the half-open interval [start date (--start-date parameter), + * end date (--end-date parameter)). --num-days parameter can be used to select data files over last --num-days. If --start-date is + * specified, --num-days will be ignored. If none of the date parameters are set, stats will be computed over all data files of all + * partitions in the table. Note that date filtering is carried out only if the partition name has the format '[column name=]yyyy-M-d', + * '[column name=]yyyy/M/d'. By default, only table level file size stats are printed. If --partition-status option is used, partition + * level file size stats also get printed. + * <br><br> + * The following stats are calculated: + * Number of files. + * Total table size. + * Minimum file size + * Maximum file size + * Average file size + * Median file size + * p50 file size + * p90 file size + * p95 file size + * p99 file size + * <br><br> + * Sample spark-submit command: + * ./bin/spark-submit \ + * --class org.apache.hudi.utilities.TableSizeStats \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.11-0.14.0-SNAPSHOT.jar \ + * --base-path <base-path> \ + * --num-days <number-of-days> + */ +public class TableSizeStats implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TableSizeStats.class); + + // 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"}; + + // Spark context + private transient JavaSparkContext jsc; + // config + private Config cfg; + // Properties with source, hoodie client, key generator etc. + private TypedProperties props; + + public TableSizeStats(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 { + @Parameter(names = {"--base-path", "-sp"}, 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 = {"--partition-stats", "-ps"}, description = "Show partition-level stats besides table-level stats.", required = false) + public boolean partitionStats = false; + + @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 "TableSizeStats {\n" + + " --base-path " + basePath + ", \n" + + " --num-days " + numDays + ", \n" + + " --start-date " + startDate + ", \n" + + " --end-date " + endDate + ", \n" + + " --partition-stats " + partitionStats + ", \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 basePath.equals(config.basePath) + && Objects.equals(numDays, config.numDays) + && Objects.equals(startDate, config.startDate) + && Objects.equals(endDate, config.endDate) + && Objects.equals(partitionStats, config.partitionStats) + && 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, partitionStats, 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); + } + + SparkConf sparkConf = UtilHelpers.buildSparkConf("Table-Size-Stats", cfg.sparkMaster); + sparkConf.set("spark.executor.memory", cfg.sparkMemory); + JavaSparkContext jsc = new JavaSparkContext(sparkConf); + + try { + TableSizeStats tableSizeStats = new TableSizeStats(jsc, cfg); + tableSizeStats.run(); + } catch (TableNotFoundException e) { + LOG.warn(String.format("The Hudi data table is not found: [%s].", cfg.basePath), e); + } catch (Throwable throwable) { + LOG.error("Failed to get table size stats for " + cfg, throwable); + } finally { + jsc.stop(); + } + } + + public void run() { + try { + LOG.info(cfg.toString()); + LOG.info(" ****** Fetching table size stats ******"); + // Set endDate to null by default. + LocalDate endDate = null; + if (cfg.endDate != null) { Review Comment: can we move the start and end date parsing to a private method and keep this run() lean. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java: ########## @@ -0,0 +1,411 @@ +/* + * 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.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.SerializableConfiguration; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.metadata.HoodieTableMetadata; + +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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaSparkContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Calculate and output file size stats of data files that were modified in the half-open interval [start date (--start-date parameter), + * end date (--end-date parameter)). --num-days parameter can be used to select data files over last --num-days. If --start-date is + * specified, --num-days will be ignored. If none of the date parameters are set, stats will be computed over all data files of all + * partitions in the table. Note that date filtering is carried out only if the partition name has the format '[column name=]yyyy-M-d', + * '[column name=]yyyy/M/d'. By default, only table level file size stats are printed. If --partition-status option is used, partition + * level file size stats also get printed. + * <br><br> + * The following stats are calculated: + * Number of files. + * Total table size. + * Minimum file size + * Maximum file size + * Average file size + * Median file size + * p50 file size + * p90 file size + * p95 file size + * p99 file size + * <br><br> + * Sample spark-submit command: + * ./bin/spark-submit \ + * --class org.apache.hudi.utilities.TableSizeStats \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.11-0.14.0-SNAPSHOT.jar \ + * --base-path <base-path> \ + * --num-days <number-of-days> + */ +public class TableSizeStats implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TableSizeStats.class); + + // 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"}; + + // Spark context + private transient JavaSparkContext jsc; + // config + private Config cfg; + // Properties with source, hoodie client, key generator etc. + private TypedProperties props; + + public TableSizeStats(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 { + @Parameter(names = {"--base-path", "-sp"}, 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 = {"--partition-stats", "-ps"}, description = "Show partition-level stats besides table-level stats.", required = false) Review Comment: --enable-partition-stats ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java: ########## @@ -0,0 +1,411 @@ +/* + * 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.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.SerializableConfiguration; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.metadata.HoodieTableMetadata; + +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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaSparkContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Calculate and output file size stats of data files that were modified in the half-open interval [start date (--start-date parameter), + * end date (--end-date parameter)). --num-days parameter can be used to select data files over last --num-days. If --start-date is + * specified, --num-days will be ignored. If none of the date parameters are set, stats will be computed over all data files of all + * partitions in the table. Note that date filtering is carried out only if the partition name has the format '[column name=]yyyy-M-d', + * '[column name=]yyyy/M/d'. By default, only table level file size stats are printed. If --partition-status option is used, partition + * level file size stats also get printed. + * <br><br> + * The following stats are calculated: + * Number of files. + * Total table size. + * Minimum file size + * Maximum file size + * Average file size + * Median file size + * p50 file size + * p90 file size + * p95 file size + * p99 file size + * <br><br> + * Sample spark-submit command: + * ./bin/spark-submit \ + * --class org.apache.hudi.utilities.TableSizeStats \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.11-0.14.0-SNAPSHOT.jar \ + * --base-path <base-path> \ + * --num-days <number-of-days> + */ +public class TableSizeStats implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TableSizeStats.class); + + // 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"}; + + // Spark context + private transient JavaSparkContext jsc; + // config + private Config cfg; + // Properties with source, hoodie client, key generator etc. + private TypedProperties props; + + public TableSizeStats(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 { + @Parameter(names = {"--base-path", "-sp"}, description = "Base path for the table", required = false) Review Comment: short form "bp" ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java: ########## @@ -0,0 +1,411 @@ +/* + * 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.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.SerializableConfiguration; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.metadata.HoodieTableMetadata; + +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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaSparkContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Calculate and output file size stats of data files that were modified in the half-open interval [start date (--start-date parameter), + * end date (--end-date parameter)). --num-days parameter can be used to select data files over last --num-days. If --start-date is + * specified, --num-days will be ignored. If none of the date parameters are set, stats will be computed over all data files of all + * partitions in the table. Note that date filtering is carried out only if the partition name has the format '[column name=]yyyy-M-d', + * '[column name=]yyyy/M/d'. By default, only table level file size stats are printed. If --partition-status option is used, partition + * level file size stats also get printed. + * <br><br> + * The following stats are calculated: + * Number of files. + * Total table size. + * Minimum file size + * Maximum file size + * Average file size + * Median file size + * p50 file size + * p90 file size + * p95 file size + * p99 file size + * <br><br> + * Sample spark-submit command: + * ./bin/spark-submit \ + * --class org.apache.hudi.utilities.TableSizeStats \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.11-0.14.0-SNAPSHOT.jar \ + * --base-path <base-path> \ + * --num-days <number-of-days> + */ +public class TableSizeStats implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TableSizeStats.class); + + // 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"}; + + // Spark context + private transient JavaSparkContext jsc; + // config + private Config cfg; + // Properties with source, hoodie client, key generator etc. + private TypedProperties props; + + public TableSizeStats(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 { + @Parameter(names = {"--base-path", "-sp"}, 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 = {"--partition-stats", "-ps"}, description = "Show partition-level stats besides table-level stats.", required = false) + public boolean partitionStats = false; + + @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 "TableSizeStats {\n" + + " --base-path " + basePath + ", \n" + + " --num-days " + numDays + ", \n" + + " --start-date " + startDate + ", \n" + + " --end-date " + endDate + ", \n" + + " --partition-stats " + partitionStats + ", \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 basePath.equals(config.basePath) + && Objects.equals(numDays, config.numDays) + && Objects.equals(startDate, config.startDate) + && Objects.equals(endDate, config.endDate) + && Objects.equals(partitionStats, config.partitionStats) + && 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, partitionStats, 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); + } + + SparkConf sparkConf = UtilHelpers.buildSparkConf("Table-Size-Stats", cfg.sparkMaster); + sparkConf.set("spark.executor.memory", cfg.sparkMemory); + JavaSparkContext jsc = new JavaSparkContext(sparkConf); + + try { + TableSizeStats tableSizeStats = new TableSizeStats(jsc, cfg); + tableSizeStats.run(); + } catch (TableNotFoundException e) { + LOG.warn(String.format("The Hudi data table is not found: [%s].", cfg.basePath), e); + } catch (Throwable throwable) { + LOG.error("Failed to get table size stats for " + cfg, throwable); + } finally { + jsc.stop(); + } + } + + public void run() { + try { + LOG.info(cfg.toString()); + LOG.info(" ****** Fetching table size stats ******"); + // Set endDate to null by default. + LocalDate endDate = null; + if (cfg.endDate != null) { + try { + endDate = LocalDate.parse(cfg.endDate, DATE_FORMATTER); + LOG.info("Setting ending date to {}. ", endDate); + } catch (DateTimeParseException dtpe) { + throw new HoodieException("Unable to parse --end-date. ", dtpe); + } + } else { + LOG.info("End date is not specified: {}.", endDate); + } + + // Set startDate to null by default. + LocalDate startDate = null; + + // Set startDate to cfg.startDate if specified. cfg.startDate takes priority over cfg.numDays if both are specified. + if (cfg.startDate != null) { + startDate = LocalDate.parse(cfg.startDate, DATE_FORMATTER); + LOG.info("Setting starting date to {}.", startDate); + } else { + if (cfg.numDays == 0) { + LOG.info("Start date not specified: {}.", startDate); + } else if (cfg.numDays > 0) { + endDate = LocalDate.now(); + startDate = endDate.minusDays(cfg.numDays); + LOG.info("Setting starting date to {} ({} - {} days). ", startDate, endDate, cfg.numDays); + } else { + throw new HoodieException("--num-days must specify a positive value."); + } + } + + // Check if starting date is before ending date. + if (startDate != null && endDate != null && !startDate.isBefore(endDate)) { + throw new HoodieException("Starting date must be before ending date. Start Date: " + startDate + ", End Date: " + endDate); + } + + if (cfg.propsFilePath != null) { + List<String> filePaths = getFilePaths(cfg.propsFilePath, jsc.hadoopConfiguration()); + for (String filePath : filePaths) { + logTableStats(filePath, startDate, endDate, cfg.partitionStats); + } + } else { + if (cfg.basePath == null) { + throw new HoodieIOException("Base path needs to be set."); + } + logTableStats(cfg.basePath, startDate, endDate, cfg.partitionStats); + } + + } catch (Exception e) { + throw new HoodieException("Unable to do fetch table size stats." + cfg.basePath, e); + } + } + + private List<String> getFilePaths(String propsPath, Configuration hadoopConf) { + List<String> filePaths = new ArrayList<>(); + FileSystem fs = FSUtils.getFs( + propsPath, + Option.ofNullable(hadoopConf).orElseGet(Configuration::new) + ); + + try (BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(new Path(propsPath))))) { + String line = reader.readLine(); + while (line != null) { + filePaths.add(line); + line = reader.readLine(); + } + } catch (IOException ioe) { + LOG.error("Error reading in properties from dfs from file." + propsPath); + throw new HoodieIOException("Cannot read properties from dfs from file " + propsPath, ioe); + } + return filePaths; + } + + private void logTableStats(String basePath, LocalDate startDate, LocalDate endDate, boolean partitionStats) throws IOException { + + LOG.warn("Processing table " + basePath); + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder() + .enable(false) + .build(); + HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc); + HoodieTableMetadata tableMetadata = HoodieTableMetadata.create(engineContext, metadataConfig, basePath, + FileSystemViewStorageConfig.SPILLABLE_DIR.defaultValue()); + SerializableConfiguration serializableConfiguration = new SerializableConfiguration(jsc.hadoopConfiguration()); + + List<String> allPartitions = tableMetadata.getAllPartitionPaths(); + + final Histogram tableHistogram = new Histogram(new UniformReservoir(1_000_000)); + allPartitions.forEach(partition -> { + // Partition name should conform to date format if startDate and/or endDate are specified. Otherwise, we don't + // need to parse partition name as date. + String dateString = partition; + if (partition.contains("=")) { + // Partition name may be in the form of "<column>=<date>". Try parsing out date. + String[] parts = partition.split("="); + dateString = parts[1].trim(); + } + + LocalDate partitionDate = null; + try { + if (startDate != null || endDate != null) { + partitionDate = LocalDate.parse(dateString, DATE_FORMATTER); + } + LOG.info("Parsed " + partition); + } catch (DateTimeParseException dtpe) { Review Comment: can we move this parsing outside of map or for each call. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java: ########## @@ -0,0 +1,411 @@ +/* + * 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.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.SerializableConfiguration; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.metadata.HoodieTableMetadata; + +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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaSparkContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Calculate and output file size stats of data files that were modified in the half-open interval [start date (--start-date parameter), Review Comment: lets start w/ high level objective of the tool -- 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]
