Pista 将 Hive 表转存至 S3 Iceberg：使用 Gravitino REST Catalog

mkdir -p pista-examples
export JAVA_HOME="/path/to/jdk-17"
export SPARK_HOME="/path/to/spark-3.5.8-bin-hadoop3"
java -version
"${SPARK_HOME}/bin/spark-submit" --version
ls "${SPARK_HOME}"/jars/hadoop-client-api-*.jar

curl -fL --retry 3 \
  -o pista-examples/pista-2.5.0.jar \
  https://github.com/jiangxt2/Pista/releases/download/v2.5.0/pista-2.5.0.jar

export PISTA_JAR="$PWD/pista-examples/pista-2.5.0.jar"

mvn -version
git clone --branch v2.5.0 --depth 1 https://github.com/jiangxt2/Pista.git pista-examples/Pista-source
mvn -B -f pista-examples/Pista-source/pom.xml package -DskipTests
export PISTA_JAR="$PWD/pista-examples/Pista-source/pista-assembly/target/pista-2.5.0.jar"

export PISTA_EXTRA_JARS="$(find -L "$PWD/pista-examples/deps" -type f -name '*.jar' | sort | paste -sd, -)"
test -n "$PISTA_EXTRA_JARS"

export PISTA_S3_ENDPOINT="https://s3.us-east-1.amazonaws.com"
export PISTA_S3_REGION="us-east-1"
export PISTA_S3_PATH_STYLE="false"
export PISTA_REST_URI="https://gravitino.example.com/iceberg"
export PISTA_HIVE_METASTORE_URI="thrift://hive-metastore.example.com:9083"
export PISTA_HIVE_DATABASE="your_source_database"
export PISTA_HIVE_TABLE="your_source_table"

export PISTA_RUN_ID="$(date -u +%Y%m%d_%H%M%S)_$$"
export PISTA_NAMESPACE="pista_example_${PISTA_RUN_ID}"

read -r -p "AWS access key ID: " AWS_ACCESS_KEY_ID
read -r -s -p "AWS secret access key: " AWS_SECRET_ACCESS_KEY
printf '\n'
export AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY
read -r -s -p "AWS session token (press Enter if unused): " AWS_SESSION_TOKEN
printf '\n'
if [[ -n "$AWS_SESSION_TOKEN" ]]; then
  export AWS_SESSION_TOKEN
else
  unset AWS_SESSION_TOKEN
fi

# pista-env.sh 中与本例相关的配置
export PISTA_ICEBERG_REST_URI="https://gravitino.example.com/iceberg"
export PISTA_S3A_ENDPOINT="https://minio.example.com"

# 本页 submit.sh 使用的变量名
export PISTA_REST_URI="${PISTA_ICEBERG_REST_URI}"
export PISTA_S3_ENDPOINT="${PISTA_S3A_ENDPOINT}"
export PISTA_S3_REGION="us-east-1"
export PISTA_S3_PATH_STYLE="true"

export PISTA_HOME="/path/to/pista"
export PISTA_CONF_DIR="$HOME/.config/pista"
source "${PISTA_HOME}/shell/load-pista-env.sh"

#!/usr/bin/env bash
set -euo pipefail

sql_name="${1:?Usage: bash pista-examples/submit.sh <name.sql> [spark options]}"
shift
[[ "$sql_name" =~ ^[a-z][a-z0-9-]*\.sql$ ]] || {
  echo "Use a plain SQL filename." >&2
  exit 1
}
sql_file="$PWD/pista-examples/$sql_name"
[[ -r "$sql_file" ]] || { echo "SQL file is not readable." >&2; exit 1; }

for key in JAVA_HOME SPARK_HOME PISTA_JAR PISTA_S3_ENDPOINT PISTA_S3_REGION PISTA_S3_PATH_STYLE PISTA_REST_URI PISTA_HIVE_METASTORE_URI PISTA_NAMESPACE PISTA_HIVE_DATABASE PISTA_HIVE_TABLE AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY; do
  [[ -n "${!key:-}" ]] || { echo "Set $key." >&2; exit 1; }
done
[[ -r "$PISTA_JAR" ]] || { echo "PISTA_JAR is not readable." >&2; exit 1; }
[[ "$PISTA_NAMESPACE" =~ ^[a-z][a-z0-9_]*$ ]] || {
  echo "Use a lowercase namespace containing letters, digits and underscores." >&2
  exit 1
}
[[ "$PISTA_HIVE_DATABASE" =~ ^[a-z][a-z0-9_]*$ ]] || {
  echo "Use a lowercase Hive database name containing letters, digits and underscores." >&2
  exit 1
}
[[ "$PISTA_HIVE_TABLE" =~ ^[a-z][a-z0-9_]*$ ]] || {
  echo "Use a lowercase Hive table name containing letters, digits and underscores." >&2
  exit 1
}
[[ "$PISTA_HIVE_METASTORE_URI" == thrift://* ]] || {
  echo "PISTA_HIVE_METASTORE_URI must use a thrift:// URI." >&2
  exit 1
}
[[ "$PISTA_S3_PATH_STYLE" == true || "$PISTA_S3_PATH_STYLE" == false ]] || {
  echo "PISTA_S3_PATH_STYLE must be true or false." >&2
  exit 1
}
[[ "$PISTA_S3_ENDPOINT" =~ ^https?:// && "$PISTA_REST_URI" =~ ^https?:// ]] || {
  echo "Use complete HTTP or HTTPS endpoint URLs." >&2
  exit 1
}
packages="org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.10.1,org.apache.iceberg:iceberg-aws-bundle:1.10.1"
spark_options=(
  --master "local[2]"
  --deploy-mode client
  --class com.pista.spark.sql.batch.SparkSQLSubmitter
  --files "$sql_file"
  --conf "spark.pista.sqlFile_=$sql_name"
  --conf spark.pista.errorPolicy=fail-fast
  --conf spark.pista.output.format=console
  --conf spark.pista.output.options.numRows=20
  --conf spark.sql.session.timeZone=UTC
  --conf spark.sql.catalogImplementation=hive
  --conf "spark.hadoop.hive.metastore.uris=$PISTA_HIVE_METASTORE_URI"
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
  --conf 'spark.redaction.regex=(?i)secret|password|token|credential|access[._-]?key'
  --conf spark.sql.catalog.lake=org.apache.iceberg.spark.SparkCatalog
  --conf spark.sql.catalog.lake.type=rest
  --conf "spark.sql.catalog.lake.uri=$PISTA_REST_URI"
  --conf spark.sql.catalog.lake.io-impl=org.apache.iceberg.aws.s3.S3FileIO
  --conf "spark.sql.catalog.lake.s3.endpoint=$PISTA_S3_ENDPOINT"
  --conf "spark.sql.catalog.lake.s3.path-style-access=$PISTA_S3_PATH_STYLE"
  --conf "spark.sql.catalog.lake.client.region=$PISTA_S3_REGION"
  --conf "spark.pista.params.namespace=$PISTA_NAMESPACE"
  --conf "spark.pista.params.hive_database=$PISTA_HIVE_DATABASE"
  --conf "spark.pista.params.hive_table=$PISTA_HIVE_TABLE"
)
if [[ -n "${PISTA_EXTRA_JARS:-}" ]]; then
  spark_options+=(--jars "$PISTA_EXTRA_JARS")
else
  spark_options+=(--packages "$packages")
fi
if [[ -n "${PISTA_SPARK_PROPERTIES:-}" ]]; then
  [[ -r "$PISTA_SPARK_PROPERTIES" ]] || {
    echo "PISTA_SPARK_PROPERTIES is not readable." >&2
    exit 1
  }
  spark_options+=(--properties-file "$PISTA_SPARK_PROPERTIES")
fi
exec "${SPARK_HOME}/bin/spark-submit" "${spark_options[@]}" "$@" "$PISTA_JAR"

SELECT id, event_date, amount, customer
FROM spark_catalog.${hive_database}.${hive_table} ORDER BY id;

SELECT COUNT(*) AS row_count, SUM(id) AS id_sum, SUM(amount) AS amount_sum
FROM spark_catalog.${hive_database}.${hive_table};

bash pista-examples/submit.sh read-hive.sql

+---+----------+------+--------+
|id |event_date|amount|customer|
+---+----------+------+--------+
|1  |2026-10-01|100.00|Alice   |
|2  |2026-10-01|250.00|Bob     |
|3  |2026-10-01|400.00|Carol   |
|4  |2026-10-02|500.00|Dave    |
+---+----------+------+--------+

+---------+------+----------+
|row_count|id_sum|amount_sum|
+---------+------+----------+
|4        |10    |1250.00   |
+---------+------+----------+

CREATE NAMESPACE IF NOT EXISTS lake.${namespace};

CREATE TABLE lake.${namespace}.events (
  id BIGINT,
  event_date DATE,
  amount DECIMAL(10, 2),
  customer STRING
)
USING iceberg
PARTITIONED BY (days(event_date));

INSERT INTO lake.${namespace}.events
SELECT id, event_date, amount, customer
FROM spark_catalog.${hive_database}.${hive_table};

WITH hive_only AS (
  SELECT id, event_date, amount, customer
  FROM spark_catalog.${hive_database}.${hive_table}
  EXCEPT ALL
  SELECT id, event_date, amount, customer FROM lake.${namespace}.events
), iceberg_only AS (
  SELECT id, event_date, amount, customer FROM lake.${namespace}.events
  EXCEPT ALL
  SELECT id, event_date, amount, customer
  FROM spark_catalog.${hive_database}.${hive_table}
)
SELECT assert_true(
  (SELECT COUNT(*) FROM hive_only) = 0 AND
  (SELECT COUNT(*) FROM iceberg_only) = 0,
  'Hive and Iceberg rows differ'
) AS verified;

bash pista-examples/submit.sh load-iceberg.sql

+--------+
|verified|
+--------+
|NULL    |
+--------+

SELECT id, event_date, amount, customer
FROM lake.${namespace}.events ORDER BY id;

WITH hive_only AS (
  SELECT id, event_date, amount, customer
  FROM spark_catalog.${hive_database}.${hive_table}
  EXCEPT ALL
  SELECT id, event_date, amount, customer FROM lake.${namespace}.events
), iceberg_only AS (
  SELECT id, event_date, amount, customer FROM lake.${namespace}.events
  EXCEPT ALL
  SELECT id, event_date, amount, customer
  FROM spark_catalog.${hive_database}.${hive_table}
)
SELECT assert_true(
  (SELECT COUNT(*) FROM hive_only) = 0 AND
  (SELECT COUNT(*) FROM iceberg_only) = 0,
  'Hive and Iceberg rows differ'
) AS verified;

SELECT snapshot_id, parent_id, operation, summary
FROM lake.${namespace}.events.snapshots ORDER BY committed_at;

bash pista-examples/submit.sh read-iceberg.sql

+---+----------+------+--------+
|id |event_date|amount|customer|
+---+----------+------+--------+
|1  |2026-10-01|100.00|Alice   |
|2  |2026-10-01|250.00|Bob     |
|3  |2026-10-01|400.00|Carol   |
|4  |2026-10-02|500.00|Dave    |
+---+----------+------+--------+

+--------+
|verified|
+--------+
|NULL    |
+--------+