diff --git a/bigtable-dataflow-parent/bigtable-beam-import/SNAPSHOT_IMPORT_USAGE.md b/bigtable-dataflow-parent/bigtable-beam-import/SNAPSHOT_IMPORT_USAGE.md new file mode 100644 index 0000000000..7e92aefe5c --- /dev/null +++ b/bigtable-dataflow-parent/bigtable-beam-import/SNAPSHOT_IMPORT_USAGE.md @@ -0,0 +1,86 @@ +# HBase Snapshot Import Helper Script Usage + +This document describes the environment variables used by the `bin/run-snapshot-import.sh` script to automate HBase snapshot imports into Cloud Bigtable using Dataflow. + +## Environment Variables + +The script relies on the following environment variables. You should set them before executing the script. + +| Variable | Status | Description | Default / Example | +| :--- | :--- | :--- | :--- | +| `PROJECT_ID` | **Required** | The Google Cloud Project ID where the Bigtable instance and Dataflow jobs reside. | `your-project-id` | +| `INSTANCE_ID` | **Required** | The Bigtable Instance ID to import data into. | `your-instance-id` | +| `BUCKET` | **Required** | The GCS bucket name. The script automatically strips any leading gs:// or trailing slashes for robust path construction. | `your-gcs-bucket` | +| `REGION` | **Required** | The GCP region to run the Dataflow jobs in. | `us-central1` | +| `TABLE_NAME` | **Required** | The target Bigtable table name. | `your-table-name` | +| `SNAPSHOT_NAME` | **Required** | The name of the HBase snapshot to import. | `your-snapshot-name` | +| `SNAPSHOT_SOURCE_DIR` | **Required** | The GCS path where the HBase snapshot export is located. | `gs://your-gcs-bucket/snapshots` | +| `NUM_SHARDS` | **Required** (except `--restore-only`) | The number of shards to split the import into for parallel processing. | `20` | +| `SERVICE_ACCOUNT` | *Optional* | The service account email to run the Dataflow jobs. | `your-service-account@developer.gserviceaccount.com` | +| `MAX_INFLIGHT_RPCS` | *Optional* | Maximum number of inflight RPCs for Bigtable client. | `100` (Default) | +| `BULK_MUTATION_CLOSE_TIMEOUT_MINUTES` | *Optional* | Timeout in minutes for closing bulk mutations. | `30` (Default) | +| `NETWORK` | *Optional* | VPC Network name for Dataflow workers. | `your-network` | +| `SUBNETWORK` | *Optional* | VPC Subnetwork name for Dataflow workers. | `regions/us-central1/subnetworks/your-subnetwork` | +| `JOB_NAME_PREFIX` | *Optional* | Safe, unique prefix for all Dataflow jobs to prevent name collisions. | `import-` (Default) | +| `WORKER_MACHINE_TYPE` | *Optional* | Compute Engine machine type for Dataflow workers. | `n1-highmem-4` (Default) | +| `DISK_SIZE_GB` | *Optional* | Worker disk size in Gigabytes. | `500` (Default) | +| `MAX_NUM_WORKERS` | *Optional* | Maximum number of active Dataflow workers per job. | `10` (Default) | +| `USE_PUBLIC_IPS` | *Optional* | Whether Dataflow workers should have public IPs. | `false` (Default) | +| `ENABLE_SNAPPY` | *Optional* | Whether to enable Snappy compression for transit files. | `true` (Default) | + +## Understanding Sharding + +An HBase snapshot consists of multiple physical **regions** (ordered chunks of table keys). To migrate extremely large tables efficiently, you can partition (shard) the regions across multiple independent Dataflow jobs. + +* **How Sharding Works:** The import tool distributes regions uniformly across a total number of shards (`NUM_SHARDS`) by hashing each region's encoded name and taking it modulo `NUM_SHARDS`. +* **Targeted Shards:** A Dataflow job configured with a specific `shardIndex` only reads and imports the HBase regions assigned to that particular shard. +* **Why use Sharding:** Instead of running a single massive Dataflow job that might hit quota limits or VM memory constraints, you can split the import into multiple smaller, parallelizable jobs. + +## Usage + +### Run a specific shard range + +Runs a range of shard indices sequentially within the script execution. + +> [!NOTE] +> Both `` and `` are **inclusive**. E.g., running `bin/run-snapshot-import.sh 0 5` will submit and execute Dataflow jobs for shards `0, 1, 2, 3, 4, and 5` sequentially. +> Shard indices must be non-negative integers strictly less than `NUM_SHARDS` (i.e. in the range `[0, NUM_SHARDS)`). The script validates this range and will fail fast if violated. + +```bash +bin/run-snapshot-import.sh +``` +Example: `bin/run-snapshot-import.sh 0 5` + +### Run all shards (Auto-parallel mode) +```bash +bin/run-snapshot-import.sh --all +``` +This mode will first run the blocking restore step, and then partition the total shards evenly and launch exactly 4 parallel runner streams in the background to process them. This ensures that no more than 4 Dataflow jobs are ever running concurrently, protecting your Bigtable instance from write-throttling and staying safely within GCP resource quotas. + +## Advanced Usage + +### Manual Parallel Execution + +To run shards in parallel groups (e.g., assuming 20 shards total), you can run multiple instances of this script. + +> [!IMPORTANT] +> Because concurrent shards cannot delete or overwrite the restored snapshot directory simultaneously, **no shard** performs the restore step during a sharded run. You MUST run the restore step explicitly first! + +Example for manual parallel execution: +```bash +# 1. Run the blocking restore step first! +bin/run-snapshot-import.sh --restore-only + +# 2. Once the restore is complete, launch shards in parallel: +bin/run-snapshot-import.sh 0 3 & +bin/run-snapshot-import.sh 4 7 & +bin/run-snapshot-import.sh 8 11 & +bin/run-snapshot-import.sh 12 15 & +bin/run-snapshot-import.sh 16 19 & +``` + +## Troubleshooting + +### JDK Compatibility + +If you are running on a newer, unsupported JDK version (e.g. JDK 25+) and hit ByteBuddy errors, you can simply set the `JVM_OPTS` variable to `"-Dnet.bytebuddy.experimental=true"` in the Configurations section of the `bin/run-snapshot-import.sh` script. diff --git a/bigtable-dataflow-parent/bigtable-beam-import/bin/run-snapshot-import.sh b/bigtable-dataflow-parent/bigtable-beam-import/bin/run-snapshot-import.sh new file mode 100755 index 0000000000..0cf938e7ae --- /dev/null +++ b/bigtable-dataflow-parent/bigtable-beam-import/bin/run-snapshot-import.sh @@ -0,0 +1,300 @@ +#!/bin/bash + +# ============================================================================== +# HBase Snapshot Import Helper Script +# ============================================================================== +# This script runs a range of Dataflow snapshot import jobs sequentially or in parallel. +# Can be run from any directory, e.g., 'bin/run-snapshot-import.sh'. +# +# For detailed usage and advanced options, see: SNAPSHOT_IMPORT_USAGE.md +# ============================================================================== + +# ------------------------------------------------------------------------------ +# Environment Variables +# ------------------------------------------------------------------------------ +# Most users will need to set these variables before running the script. +# See SNAPSHOT_IMPORT_USAGE.md for details and expected values. + +# --- Required / Common Configurations --- +# export PROJECT_ID="your-project-id" +# export INSTANCE_ID="your-instance-id" +# export BUCKET="your-gcs-bucket" +# export REGION="us-central1" +# +# export TABLE_NAME="your-table-name" +# export SNAPSHOT_NAME="your-snapshot-name" +# export SNAPSHOT_SOURCE_DIR="gs://your-gcs-bucket/snapshots" +# export SERVICE_ACCOUNT="your-service-account" + +# --- Sharding & Tuning --- +# A 'shard' is a partition of HBase regions. Sharding allows splitting the import into +# multiple independent Dataflow jobs running in parallel to speed up the migration. +# export NUM_SHARDS="20" +# export MAX_INFLIGHT_RPCS="100" +# export BULK_MUTATION_CLOSE_TIMEOUT_MINUTES="30" + +# --- Network Configurations --- +# export NETWORK="your-network" +# export SUBNETWORK="your-subnetwork" + +# ------------------------------------------------------------------------------ +# Usage +# ------------------------------------------------------------------------------ +# Usage: bin/run-snapshot-import.sh +# (Both start and end shard indices are inclusive) +# Or: bin/run-snapshot-import.sh --all +# (Runs all shards in parallel groups of 4 by default) +# +# Examples: +# bin/run-snapshot-import.sh 0 3 +# bin/run-snapshot-import.sh --all + +# Validate and parse arguments +if [ "$1" != "--all" ] && [ "$1" != "--restore-only" ]; then + if [ -z "$1" ] || [ -z "$2" ]; then + echo "Usage: $0 [additional_args...]" + echo " (Both start and end shard indices are inclusive)" + echo " Or: $0 --all [additional_args...]" + echo " Or: $0 --restore-only [additional_args...]" + exit 1 + fi + if ! [[ "$1" =~ ^[0-9]+$ ]] || ! [[ "$2" =~ ^[0-9]+$ ]]; then + echo "❌ Error: Shard indices must be non-negative integers." + exit 1 + fi + if [ "$1" -gt "$2" ]; then + echo "❌ Error: start_shard ($1) cannot be greater than end_shard ($2)." + exit 1 + fi + START_SHARD=$1 + END_SHARD=$2 + EXTRA_ARGS=("${@:3}") +else + EXTRA_ARGS=("${@:2}") +fi + +# Validate and export required environment variables +REQUIRED_VARS=(PROJECT_ID INSTANCE_ID BUCKET REGION TABLE_NAME SNAPSHOT_NAME SNAPSHOT_SOURCE_DIR) +for var in "${REQUIRED_VARS[@]}"; do + if [ -z "${!var}" ]; then + echo "❌ Error: Environment variable $var is not set." + exit 1 + fi + export "${var}" +done + +if [ "$1" != "--restore-only" ]; then + if [ -z "${NUM_SHARDS}" ]; then + echo "❌ Error: Environment variable NUM_SHARDS is not set." + exit 1 + fi + if ! [[ "${NUM_SHARDS}" =~ ^[0-9]+$ ]] || [ "${NUM_SHARDS}" -le 0 ]; then + echo "❌ Error: NUM_SHARDS must be a positive integer." + exit 1 + fi + export NUM_SHARDS + if [ "$1" != "--all" ]; then + if [ "${START_SHARD}" -ge "${NUM_SHARDS}" ] || [ "${END_SHARD}" -ge "${NUM_SHARDS}" ]; then + echo "❌ Error: Shard indices (${START_SHARD}, ${END_SHARD}) must be less than NUM_SHARDS (${NUM_SHARDS})." + exit 1 + fi + fi +fi + +# Strip leading gs:// and trailing slashes from BUCKET for robust GCS path construction +BUCKET="${BUCKET#gs://}" +BUCKET="${BUCKET%/}" + +# Set default values for optional tuning parameters +MAX_INFLIGHT_RPCS="${MAX_INFLIGHT_RPCS:-100}" +BULK_MUTATION_CLOSE_TIMEOUT_MINUTES="${BULK_MUTATION_CLOSE_TIMEOUT_MINUTES:-30}" + +# Configurable Dataflow execution parameters +WORKER_MACHINE_TYPE="${WORKER_MACHINE_TYPE:-n1-highmem-4}" +DISK_SIZE_GB="${DISK_SIZE_GB:-500}" +MAX_NUM_WORKERS="${MAX_NUM_WORKERS:-10}" +USE_PUBLIC_IPS="${USE_PUBLIC_IPS:-false}" +ENABLE_SNAPPY="${ENABLE_SNAPPY:-true}" + +# Generate a safe, unique job name prefix to prevent collisions +SAFE_TABLE_NAME=$(echo "${TABLE_NAME}" | tr '[:upper:]' '[:lower:]' | tr '_' '-' | tr -cd '[:alnum:]-') +JOB_NAME_PREFIX="${JOB_NAME_PREFIX:-import-${SAFE_TABLE_NAME}}" + +# Configurations +SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &> /dev/null && pwd) + +# Detect the newest shaded JAR dynamically in the target directory +JAR_PATH="" +for jar in "${SCRIPT_DIR}"/../target/bigtable-beam-import-*-shaded.jar; do + if [ -f "$jar" ]; then + if [ -z "${JAR_PATH}" ] || [ "$jar" -nt "${JAR_PATH}" ]; then + JAR_PATH="$jar" + fi + fi +done + +if [ -z "${JAR_PATH}" ]; then + echo "❌ Error: Shaded JAR not found in ${SCRIPT_DIR}/../target/" + echo " Please compile and package the project first by running:" + echo " mvn clean package -DskipTests" + exit 1 +fi +RESTORE_DIR="gs://${BUCKET}/restore-${SNAPSHOT_NAME}" +# If running the launcher on a newer, unsupported JDK version (e.g. JDK 25+), +# you can set JVM_OPTS to "-Dnet.bytebuddy.experimental=true" to bypass +# ByteBuddy class generation crashes during pipeline construction. +JVM_OPTS="${JVM_OPTS:-}" + +# Construct optional network arguments +NETWORK_ARGS=() +if [ -n "${NETWORK}" ]; then + NETWORK_ARGS+=("--network=${NETWORK}") +fi +if [ -n "${SUBNETWORK}" ]; then + NETWORK_ARGS+=("--subnetwork=${SUBNETWORK}") +fi + +# Construct optional service account arguments +SERVICE_ACCOUNT_ARGS=() +if [ -n "${SERVICE_ACCOUNT}" ]; then + SERVICE_ACCOUNT_ARGS+=("--serviceAccount=${SERVICE_ACCOUNT}") +fi + + +# --- RESTORE ONLY MODE --- +if [ "$1" == "--restore-only" ]; then + echo "🚀 Performing snapshot restore (blocking)..." + if ! java ${JVM_OPTS} -jar "${JAR_PATH}" importsnapshot \ + --runner=DataflowRunner \ + --project="${PROJECT_ID}" \ + --bigtableInstanceId="${INSTANCE_ID}" \ + --bigtableTableId="${TABLE_NAME}" \ + --hbaseSnapshotSourceDir="${SNAPSHOT_SOURCE_DIR}" \ + --snapshots="${SNAPSHOT_NAME}:${TABLE_NAME}" \ + --stagingLocation="gs://${BUCKET}/dataflow/staging" \ + --tempLocation="gs://${BUCKET}/dataflow/temp" \ + --region="${REGION}" \ + --performOnlyRestoreStep=true \ + --restorePath="${RESTORE_DIR}" \ + --jobName="${JOB_NAME_PREFIX}-restore" \ + "${SERVICE_ACCOUNT_ARGS[@]}" \ + "${NETWORK_ARGS[@]}" \ + "${EXTRA_ARGS[@]}"; then + echo "❌ Error: Snapshot restore failed." + exit 1 + fi + echo "✅ Restore completed." + echo "⚠️ IMPORTANT: Please manually cleanup the restore path once validation succeeds:" + echo " gsutil rm -r \"${RESTORE_DIR}\"" + exit 0 +fi + +# --- AUTO-PARALLEL MODE --- +if [ "$1" == "--all" ]; then + echo "🚀 Starting fully automated snapshot import..." + + # Step 1: Perform ONLY the restore step + echo "Step 1/2: Performing snapshot restore (blocking)..." + if ! java ${JVM_OPTS} -jar "${JAR_PATH}" importsnapshot \ + --runner=DataflowRunner \ + --project="${PROJECT_ID}" \ + --bigtableInstanceId="${INSTANCE_ID}" \ + --bigtableTableId="${TABLE_NAME}" \ + --hbaseSnapshotSourceDir="${SNAPSHOT_SOURCE_DIR}" \ + --snapshots="${SNAPSHOT_NAME}:${TABLE_NAME}" \ + --stagingLocation="gs://${BUCKET}/dataflow/staging" \ + --tempLocation="gs://${BUCKET}/dataflow/temp" \ + --region="${REGION}" \ + --performOnlyRestoreStep=true \ + --restorePath="${RESTORE_DIR}" \ + --jobName="${JOB_NAME_PREFIX}-restore" \ + "${SERVICE_ACCOUNT_ARGS[@]}" \ + "${NETWORK_ARGS[@]}" \ + "${EXTRA_ARGS[@]}"; then + echo "❌ Error: Snapshot restore failed. Aborting import." + exit 1 + fi + + echo "Restore completed. Proceeding to data import." + + # Step 2: Launch 4 parallel runner streams + echo "Step 2/2: Launching 4 parallel runner streams..." + MAX_CONCURRENCY=4 + SHARDS_PER_RUNNER=$(( (NUM_SHARDS + MAX_CONCURRENCY - 1) / MAX_CONCURRENCY )) + + pids=() + trap 'echo "⚠️ Interrupted. Terminating background runners..."; [ ${#pids[@]} -gt 0 ] && kill "${pids[@]}" 2>/dev/null; exit 1' INT TERM + for (( runner=0; runnerMETA-INF/*.RSA META-INF/**/pom.properties META-INF/**/pom.xml - META-INF/MANIFEST.MF META-INF/LICENSE META-INF/NOTICE.txt META-INF/NOTICE