Skip to main content

Table Maintenance

Overview​

The table maintenance service keeps tables healthy without anyone watching them. You attach a policy to a catalog, schema, or table; the service collects statistics, evaluates them against that policy, and submits a job when the policy says work is needed.

The framework is generic. Metrics collection, policy evaluation, and job submission are not tied to any particular table format, and each is a Java ServiceLoader extension point. What ships built in is deliberately narrower, and in alpha that means Iceberg data file compaction on identity-partitioned tables.

The CLI binary, its configuration file, and its configuration keys carry the older name optimizer, so you will see gravitino-optimizer.sh, gravitino-optimizer.conf, and gravitino.optimizer.* throughout. Those are literal strings rather than a second product.

Alpha Scope​

Confirm your environment matches this list before starting an evaluation against the built-ins. Anything outside it needs a custom extension, which is covered in the Extension Guide.

  • Compaction is the only built-in strategy. There is no built-in snapshot expiration, orphan file cleanup, or sort and cluster maintenance.
  • Compaction applies to Iceberg tables only, and only where every partition uses an identity transform.
  • The service is driven through the CLI workflow rather than running on a schedule of its own.

How It Works​

Maintenance runs as four steps. Each is a separate command, so you can stop after any of them, and the dry run on step two shows what would be submitted before anything runs.

StepWhat you runWhat it produces
Collectupdate-statistics, append-metricsStatistics on the table, metrics in the JDBC repository
Evaluatesubmit-strategy-jobs --dry-runCandidate actions, with nothing submitted
Submitsubmit-strategy-jobs, submit-update-stats-jobA Spark job, tracked by job status and staging logs
Observemonitor-metrics, list-table-metricsBefore and after metrics, and rewritten data files

Naming​

Three identifiers look interchangeable and are not.

TermExampleWhere it appears
Policy nameiceberg_compaction_defaultThe policy's own name, and the CLI --strategy-name
Policy typesystem_iceberg_compactionThe policyType field when creating a policy over REST
Strategy typeiceberg-data-compactionThe strategy.type field, and the strategy handler config

--strategy-name takes the policy name, despite what it is called. Passing either of the other two reports no matching identifiers rather than naming the mistake.

Configuration​

Three layers of configuration apply, and they are set in different places for different lifetimes. Server configuration governs how jobs run at all, CLI configuration governs how the commands reach Gravitino, and jobConf governs a single job submission.

LayerWhere it livesLifetime
Servergravitino.confUntil server restart
CLIconf/gravitino-optimizer.confPer command
Job submissionjobConf in the request bodyOne job run

Server Configuration​

Set these in gravitino.conf. They control the job executor rather than maintenance itself, so they apply to every job Gravitino runs.

gravitino.job.executor=local
gravitino.job.statusPullIntervalInMs=300000
gravitino.jobExecutor.local.sparkHome=/path/to/spark

gravitino.job.statusPullIntervalInMs defaults to five minutes. Job status is polled rather than pushed, so REST status can lag the real Spark process by a full interval, which makes a working job look hung. Lower it to 10000 for local work and restart the server.

CLI Configuration​

The CLI needs to know where Gravitino is and which components to use. This is a minimal working file for submit-strategy-jobs.

gravitino.optimizer.gravitinoUri = http://localhost:8090
gravitino.optimizer.gravitinoMetalake = test
gravitino.optimizer.gravitinoDefaultCatalog = rest_catalog
gravitino.optimizer.recommender.statisticsProvider = gravitino-statistics-provider
gravitino.optimizer.recommender.strategyProvider = gravitino-strategy-provider
gravitino.optimizer.recommender.tableMetaProvider = gravitino-table-metadata-provider
gravitino.optimizer.recommender.jobSubmitter = gravitino-job-submitter
gravitino.optimizer.strategyHandler.iceberg-data-compaction.className = org.apache.gravitino.maintenance.optimizer.recommender.handler.compaction.CompactionStrategyHandler
gravitino.optimizer.jobSubmitterConfig.catalog_name = rest_catalog
gravitino.optimizer.jobSubmitterConfig.spark_master = local[2]
gravitino.optimizer.jobSubmitterConfig.spark_executor_instances = 1
gravitino.optimizer.jobSubmitterConfig.spark_executor_cores = 1
gravitino.optimizer.jobSubmitterConfig.spark_executor_memory = 1g
gravitino.optimizer.jobSubmitterConfig.spark_driver_memory = 1g
gravitino.optimizer.jobSubmitterConfig.catalog_type = rest
gravitino.optimizer.jobSubmitterConfig.catalog_uri = http://localhost:9001/iceberg
# Leave empty for a local filesystem; set to your warehouse URI for cloud or HDFS storage.
gravitino.optimizer.jobSubmitterConfig.warehouse_location =
gravitino.optimizer.jobSubmitterConfig.spark_conf = {"spark.master":"local[2]","spark.hadoop.fs.defaultFS":"file:///"}

Everything under gravitino.optimizer.jobSubmitterConfig. becomes the jobConf of jobs this CLI submits, so the two layers carry the same keys under different names.

Job Submission Configuration​

A direct job submission carries its own jobConf. This is builtin-iceberg-update-stats with the keys it needs.

{
  "catalog_name": "rest_catalog",
  "table_identifier": "db.t1",
  "update_mode": "all",
  "updater_options": "{\"gravitino_uri\":\"http://localhost:8090\",\"metalake\":\"test\",\"statistics_updater\":\"gravitino-statistics-updater\",\"metrics_updater\":\"gravitino-metrics-updater\"}",
  "spark_conf": "{\"spark.master\":\"local[2]\",\"spark.hadoop.fs.defaultFS\":\"file:///\"}",
  "spark_master": "local[2]",
  "spark_executor_instances": "1",
  "spark_executor_cores": "1",
  "spark_executor_memory": "1g",
  "spark_driver_memory": "1g",
  "catalog_type": "rest",
  "catalog_uri": "http://localhost:9001/iceberg",
  "warehouse_location": ""
}

updater_options and spark_conf are JSON strings inside a JSON object, so their quotes are escaped. That nesting is the most common source of malformed submissions.

warehouse_location may be empty for local filesystem testing. Set it to the warehouse URI for HDFS or cloud object storage.

Running Against a Local Filesystem​

On a machine with no HDFS, Spark still defaults to hdfs://localhost:9000 and fails. Set the default filesystem explicitly, in spark_conf for job submissions and in the CLI spark_conf value:

spark.hadoop.fs.defaultFS=file:///

Checking Your Configuration​

Four things are worth confirming before assuming a configuration problem is a code problem.

  • builtin-iceberg-update-stats and builtin-iceberg-rewrite-data-files appear in the job template list.
  • The policy is attached to the target table, not merely created.
  • submit-strategy-jobs prints SUBMIT lines rather than nothing.
  • The rewrite log shows Rewritten data files: N with N greater than zero for a non-empty table.

Walkthrough​

This takes one Iceberg table through the whole workflow: create it, fill it with small files, attach a compaction policy, collect statistics, and let the service decide to compact it. It runs against a local Spark and takes about fifteen minutes.

Each step ends with a check. If a check fails, stop there, since every step depends on the one before it.

Prerequisites​

  • A running Gravitino server with a metalake. The examples use test.
  • Spark available to the job executor, through either SPARK_HOME or gravitino.jobExecutor.local.sparkHome.
  • gravitino.job.statusPullIntervalInMs lowered to 10000 and the server restarted. The default is five minutes, which makes every status check in this walkthrough feel broken.

If your Iceberg REST backend runs in memory, do not restart it partway through. Restarting resets both metadata and data files, and you start over.

Step 1: Confirm the Job Templates Exist​

curl -sS "http://localhost:8090/api/metalakes/test" | jq
curl -sS "http://localhost:8090/api/metalakes/test/jobs/templates?details=true" \
  | jq '.jobTemplates[].name'

The template list must include builtin-iceberg-update-stats and builtin-iceberg-rewrite-data-files. If it does not, the gravitino-jobs JAR is missing from auxlib. Add it and restart the server before going on.

Step 2: Create the Demo Catalog, Schema, and Table​

# Catalog. An "already exists" error here is fine.
curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "rest_catalog",
    "type": "RELATIONAL",
    "comment": "Iceberg REST catalog",
    "provider": "lakehouse-iceberg",
    "properties": {
      "catalog-backend": "rest",
      "uri": "http://localhost:9001/iceberg"
    }
  }' \
  http://localhost:8090/api/metalakes/test/catalogs

# Schema
curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" \
  -d '{"name": "db", "comment": "maintenance demo schema", "properties": {}}' \
  http://localhost:8090/api/metalakes/test/catalogs/rest_catalog/schemas

# Table
curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "t1",
    "comment": "maintenance demo table",
    "columns": [
      {"name": "id", "type": "integer", "nullable": true},
      {"name": "name", "type": "string", "nullable": true}
    ],
    "properties": {}
  }' \
  http://localhost:8090/api/metalakes/test/catalogs/rest_catalog/schemas/db/tables

Step 3: Create Something Worth Compacting​

An empty table gives the policy nothing to react to, so write 100,000 rows capped at 1,000 rows per file. That produces the many small files compaction exists to merge.

${SPARK_HOME}/bin/spark-sql \
  --conf spark.hadoop.fs.defaultFS=file:/// \
  --conf spark.sql.catalog.rest_catalog=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.rest_catalog.type=rest \
  --conf spark.sql.catalog.rest_catalog.uri=http://localhost:9001/iceberg \
  -e "CREATE NAMESPACE IF NOT EXISTS rest_catalog.db; \
      SET spark.sql.files.maxRecordsPerFile=1000; \
      INSERT INTO rest_catalog.db.t1 \
      SELECT id, concat('name_', CAST(id AS STRING)) FROM range(0, 100000);"

Without spark.hadoop.fs.defaultFS=file:///, Spark reaches for hdfs://localhost:9000 and fails.

Step 4: Attach a Compaction Policy​

Creating the policy is not enough. It has to be attached to the table, and the attachment is what the service reads.

curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "iceberg_compaction_default",
    "comment": "Built-in Iceberg compaction policy",
    "policyType": "system_iceberg_compaction",
    "enabled": true,
    "content": {}
  }' \
  http://localhost:8090/api/metalakes/test/policies

curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" \
  -d '{"policiesToAdd": ["iceberg_compaction_default"]}' \
  http://localhost:8090/api/metalakes/test/objects/table/rest_catalog.db.t1/policies

Confirm the attachment before moving on:

curl -sS "http://localhost:8090/api/metalakes/test/objects/table/rest_catalog.db.t1/policies?details=true" | jq

Step 5: Collect Statistics​

update_stats_job_id=$(curl -sS -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" \
  -d '{
    "jobTemplateName": "builtin-iceberg-update-stats",
    "jobConf": {
      "catalog_name": "rest_catalog",
      "table_identifier": "db.t1",
      "update_mode": "all",
      "updater_options": "{\"gravitino_uri\":\"http://localhost:8090\",\"metalake\":\"test\",\"statistics_updater\":\"gravitino-statistics-updater\",\"metrics_updater\":\"gravitino-metrics-updater\"}",
      "spark_conf": "{\"spark.master\":\"local[2]\",\"spark.hadoop.fs.defaultFS\":\"file:///\"}",
      "spark_master": "local[2]",
      "spark_executor_instances": "1",
      "spark_executor_cores": "1",
      "spark_executor_memory": "1g",
      "spark_driver_memory": "1g",
      "catalog_type": "rest",
      "catalog_uri": "http://localhost:9001/iceberg",
      "warehouse_location": ""
    }
  }' \
  http://localhost:8090/api/metalakes/test/jobs/runs | jq -r '.job.jobId')

echo "update-stats job id: ${update_stats_job_id}"

Wait for it to finish, then confirm the statistics landed:

curl -sS "http://localhost:8090/api/metalakes/test/objects/table/rest_catalog.db.t1/statistics" | jq

The response must include custom-data-file-mse and custom-delete-file-number. Those two are what the compaction policy evaluates, so if they are absent the next step has nothing to decide on.

Step 6: Evaluate and Submit​

Write the CLI configuration first. --strategy-name takes the policy name, not the policy type or the strategy type.

cat > /tmp/gravitino-optimizer-submit.conf <<'EOF_CONF'
gravitino.optimizer.gravitinoUri = http://localhost:8090
gravitino.optimizer.gravitinoMetalake = test
gravitino.optimizer.gravitinoDefaultCatalog = rest_catalog
gravitino.optimizer.recommender.statisticsProvider = gravitino-statistics-provider
gravitino.optimizer.recommender.strategyProvider = gravitino-strategy-provider
gravitino.optimizer.recommender.tableMetaProvider = gravitino-table-metadata-provider
gravitino.optimizer.recommender.jobSubmitter = gravitino-job-submitter
gravitino.optimizer.strategyHandler.iceberg-data-compaction.className = org.apache.gravitino.maintenance.optimizer.recommender.handler.compaction.CompactionStrategyHandler
gravitino.optimizer.jobSubmitterConfig.catalog_name = rest_catalog
gravitino.optimizer.jobSubmitterConfig.spark_master = local[2]
gravitino.optimizer.jobSubmitterConfig.spark_executor_instances = 1
gravitino.optimizer.jobSubmitterConfig.spark_executor_cores = 1
gravitino.optimizer.jobSubmitterConfig.spark_executor_memory = 1g
gravitino.optimizer.jobSubmitterConfig.spark_driver_memory = 1g
gravitino.optimizer.jobSubmitterConfig.catalog_type = rest
gravitino.optimizer.jobSubmitterConfig.catalog_uri = http://localhost:9001/iceberg
# Leave empty for a local filesystem; set to your warehouse URI for cloud or HDFS storage.
gravitino.optimizer.jobSubmitterConfig.warehouse_location =
gravitino.optimizer.jobSubmitterConfig.spark_conf = {"spark.master":"local[2]","spark.hadoop.fs.defaultFS":"file:///"}
EOF_CONF

Preview first. A dry run evaluates the policy and prints what it would do without submitting anything.

./bin/gravitino-optimizer.sh \
  --type submit-strategy-jobs \
  --identifiers rest_catalog.db.t1 \
  --strategy-name iceberg_compaction_default \
  --dry-run \
  --limit 10 \
  --conf-path /tmp/gravitino-optimizer-submit.conf

DRY-RUN lines mean the policy fired. No output at all means it did not, which usually means the statistics from step 5 are below the policy thresholds rather than that anything is broken.

Then submit for real:

submit_output=$(./bin/gravitino-optimizer.sh \
  --type submit-strategy-jobs \
  --identifiers rest_catalog.db.t1 \
  --strategy-name iceberg_compaction_default \
  --limit 10 \
  --conf-path /tmp/gravitino-optimizer-submit.conf)
echo "${submit_output}"

strategy_job_id=$(echo "${submit_output}" | sed -n 's/.*jobId=\([^[:space:]]*\).*/\1/p')
[[ -z "${strategy_job_id}" ]] && echo 'ERROR: failed to extract strategy job ID' && exit 1
echo "strategy rewrite job id: ${strategy_job_id}"

Step 7: Verify the Rewrite​

curl -sS "http://localhost:8090/api/metalakes/test/jobs/runs/${update_stats_job_id}" | jq
curl -sS "http://localhost:8090/api/metalakes/test/jobs/runs/${strategy_job_id}" | jq

log_dir="/tmp/gravitino/jobs/staging/test/builtin-iceberg-rewrite-data-files/${strategy_job_id}"
grep -E "Rewritten data files|Added data files|completed successfully" "${log_dir}/output.log"

Rewritten data files: N with N greater than zero means the workflow worked end to end. The staging path comes from gravitino.job.stagingDir, which defaults to /tmp/gravitino/jobs/staging.

REST job status is polled rather than pushed, so it lags the real Spark process by up to one poll interval. That is why the prerequisites lower it to ten seconds.

Last updated on