Microsegment.ru
  • Главная страница
  • О проекте
  • Портфолио
  • Блог
Системы

Автоматизация наполнения данными гибридного хранилища данных

Автоматизация наполнения данными гибридного хранилища данных
Системы

Здесь представлены файлы bash для автоматизации чтения данных из MS SQL Server в слой STG, а далее из слоя STG в квартальные партиции слоя RAW минимального жизнеспособного (MVP) гибридного хранилища данных (Data Lakehouse) на основе PostgreSQL.

Содержание:

  1. Скрипт файла mssql_to_stg.sh для чтения данных из MS SQL Server в слой STG гибридного хранилища данных
  2. Скрипт файла stg_to_raw.sh для чтения данных из слоя STG в слой RAW

Скрипт файла mssql_to_stg.sh для чтения данных из MS SQL Server в слой STG гибридного хранилища данных¶

Файл mssql_to_stg.sh — предназначен для автоматизации чтения данных из MS SQL Server в слой STG Data Lakehouse.

mssql_to_stg.sh

#!/usr/bin/env bash
# ============================================================================
# mssql_to_stg.sh — ETL: MS SQL Server → STG → RAW (Data Lakehouse MVP)
#
# Назначение (согласно архитектуре DLH MVP):
#   * STG — «песочница», полностью перезаписывается при каждой загрузке,
#     хранит данные 30 суток, таблица без индексов.
#   * RAW — долгосрочное неизменяемое хранилище, партиционированное по
#     кварталам, каждая партиция — в отдельном tablespace на LVM.
#
# Принципы (те же, что в create_raw.sh / drop_raw.sh / archive / restore):
#   1. Идемпотентность: повторный запуск не создаёт дублей в RAW.
#   2. Атомарность: STG перезаписывается в одной транзакции, RAW — через
#      ON CONFLICT (id) DO NOTHING.
#   3. Наблюдаемость: предполётный отчёт + верификация STG ↔ RAW.
#   4. Готовность к CI/CD: --dry-run, вынесенная конфигурация, flock.
#   5. Non-disruptive: PostgreSQL не останавливается.
#
# Зависимости:
#   * Python 3 + pyodbc + psycopg2 (в закрытом контуре — заранее загружены).
#   * Вспомогательный скрипт extract_mssql.py (см. раздел «Компоненты»).
#   * create_new_partition.sh в PATH или в /opt/dlh/scripts/.
#
# Использование:
#   sudo ./mssql_to_stg.sh                    # полная перезагрузка STG → RAW
#   sudo ./mssql_to_stg.sh --since=2026-01-01 # инкремент по dt
#   sudo ./mssql_to_stg.sh --dry-run          # показать план, ничего не менять
# ============================================================================

set -Eeuo pipefail
trap 'on_error $? $LINENO "${BASH_COMMAND}"' ERR
trap 'cleanup' EXIT

on_error() {
    log "КРИТИЧЕСКАЯ ОШИБКА: код=$1, строка=$2, команда='$3'"
    exit "$1"
}

cleanup() {
    # Удаляем временные CSV/файлы, если остались
    [[ -n "${TMP_DIR:-}" && -d "${TMP_DIR:-}" ]] && rm -rf "$TMP_DIR" 2>/dev/null || true
    log "Завершение stg_to_raw.sh."
}

# ---------------------------------------------------------------------------
# 1. Конфигурация
# ---------------------------------------------------------------------------
CONFIG_FILE="${CONFIG_FILE:-/etc/dlh/raw.env}"
[[ -f "$CONFIG_FILE" ]] && source "$CONFIG_FILE"

PG_VERSION="${PG_VERSION:-17}"
PG_SERVICE="${PG_SERVICE:-postgresql-${PG_VERSION}}"
PG_DB="${PG_DB:-dlh_db}"
PG_USER="${PG_USER:-postgres}"

RAW_SCHEMA="${RAW_SCHEMA:-raw}"
STG_SCHEMA="${STG_SCHEMA:-stg}"
TABLE_NAME="${TABLE_NAME:-table_1}"
STG_TABLE="${STG_SCHEMA}.${TABLE_NAME}"
RAW_TABLE="${RAW_SCHEMA}.${TABLE_NAME}"

# Источник MS SQL Server
MSSQL_HOST="${MSSQL_HOST:-mssql.source.local}"
MSSQL_PORT="${MSSQL_PORT:-1433}"
MSSQL_DB="${MSSQL_DB:-source_db}"
MSSQL_USER="${MSSQL_USER:-etl_reader}"
MSSQL_PASSWORD="${MSSQL_PASSWORD:-}"          # лучше — из secret store
MSSQL_TABLE="${MSSQL_TABLE:-dbo.Table1}"

# Роли (least privilege)
APP_USER="${APP_USER:-dlh_user}"

# Вспомогательные пути
PYTHON_EXTRACTOR="${PYTHON_EXTRACTOR:-/opt/dlh/scripts/extract_mssql.py}"
CREATE_PARTITION_SCRIPT="${CREATE_PARTITION_SCRIPT:-/opt/dlh/scripts/create_new_partition.sh}"

LOG_DIR="${LOG_DIR:-/var/log/dlh/scripts}"
LOG_FILE="${LOG_FILE:-${LOG_DIR}/stg_to_raw_$(date +%Y%m%d_%H%M%S).log}"

# Таймауты
LOCK_TIMEOUT="${LOCK_TIMEOUT:-10s}"
STATEMENT_TIMEOUT="${STATEMENT_TIMEOUT:-300s}"   # 5 минут на COPY/INSERT

# ---------------------------------------------------------------------------
# 2. Аргументы
# ---------------------------------------------------------------------------
SINCE=""
DRY_RUN=false
FORCE_FULL=false

usage() {
    cat <<EOF
Использование: $0 [опции]

Опции:
  --since=YYYY-MM-DD    Инкрементальная загрузка: только строки с dt >= указанной даты.
                        Без опции — полная перезагрузка STG и загрузка в RAW.
  --full-refresh        Принудительно полная перезагрузка (игнорирует --since).
  --dry-run             Показать план, ничего не выполнять.
  -h, --help            Эта справка.

Конфигурация: $CONFIG_FILE (или CONFIG_FILE=... в окружении).
EOF
}

while [[ $# -gt 0 ]]; do
    case "$1" in
        --since=*)       SINCE="${1#*=}" ;;
        --full-refresh)  FORCE_FULL=true ;;
        --dry-run)       DRY_RUN=true ;;
        -h|--help)       usage; exit 0 ;;
        *)               echo "Неизвестная опция: $1" >&2; usage; exit 2 ;;
    esac
    shift
done

if [[ -n "$SINCE" ]] && ! [[ "$SINCE" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]]; then
    echo "ОШИБКА: --since должен быть в формате YYYY-MM-DD: ${SINCE}" >&2
    exit 2
fi
$FORCE_FULL && SINCE=""

# ---------------------------------------------------------------------------
# 3. Базовые функции
# ---------------------------------------------------------------------------
mkdir -p "$LOG_DIR"

log() {
    local msg; msg="$(date '+%Y-%m-%d %H:%M:%S') - $1"
    echo "$msg" | tee -a "$LOG_FILE" >&2
}

die() { log "ФАТАЛЬНАЯ ОШИБКА: $1"; exit 1; }

require_root() { [[ $EUID -eq 0 ]] || die "Требуется root (sudo)."; }

acquire_lock() {
    exec 9>"/var/lock/stg_to_raw.lock"
    flock -n 9 || die "Другой экземпляр stg_to_raw.sh уже выполняется."
}

run_cmd() {
    local desc="$1"; shift
    if $DRY_RUN; then log "[DRY-RUN] ${desc}"; return 0; fi
    log "${desc}"; "$@"
}

run_sql() {
    local sql="$1"; shift
    if $DRY_RUN; then
        log "[DRY-RUN] SQL: $(printf '%s' "$sql" | tr '\n' ' ' | head -c 400)"
        return 0
    fi
    log "SQL (${#sql} байт)."
    printf '%s\n' "$sql" |
        sudo -u "$PG_USER" -- psql -X -q -d "$PG_DB" \
            -v ON_ERROR_STOP=1 -1 "$@" -f - >/dev/null \
        || { log "Ошибка SQL."; return 1; }
}

psql_at() {
    sudo -u "$PG_USER" -- psql -X -At -d "$PG_DB" -v ON_ERROR_STOP=1 -c "$1" 2>/dev/null || true
}

# ---------------------------------------------------------------------------
# 4. Проверка зависимостей
# ---------------------------------------------------------------------------
check_dependencies() {
    log "--- Проверка зависимостей ---"

    # PostgreSQL
    if ! systemctl is-active --quiet "$PG_SERVICE"; then
        run_cmd "Запуск ${PG_SERVICE}" systemctl start "$PG_SERVICE" \
            || die "Не удалось запустить ${PG_SERVICE}."
    fi
    for _ in {1..30}; do
        sudo -u "$PG_USER" -- pg_isready -q && break
        sleep 1
    done
    sudo -u "$PG_USER" -- pg_isready -q || die "PostgreSQL не готов."

    # Python-экстрактор
    if [[ ! -f "$PYTHON_EXTRACTOR" ]]; then
        die "Python-экстрактор не найден: ${PYTHON_EXTRACTOR}. См. раздел «Компоненты»."
    fi

    # Скрипт создания партиций
    if [[ ! -x "$CREATE_PARTITION_SCRIPT" ]]; then
        log "ВНИМАНИЕ: ${CREATE_PARTITION_SCRIPT} не найден или не исполняем. Автосоздание партиций недоступно."
    fi

    # Проверка, что STG-схема и таблица существуют
    local stg_exists
    stg_exists="$(psql_at "SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='${STG_SCHEMA}' AND c.relname='${TABLE_NAME}';")"
    [[ "$stg_exists" == "1" ]] || die "STG-таблица ${STG_TABLE} не найдена. Создайте её (например, через create_raw.sh)."

    local raw_exists
    raw_exists="$(psql_at "SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='${RAW_SCHEMA}' AND c.relname='${TABLE_NAME}' AND c.relkind='p';")"
    [[ "$raw_exists" == "1" ]] || die "RAW-таблица ${RAW_TABLE} не найдена или не является партиционированной."

    log "Все зависимости в порядке."
}

# ---------------------------------------------------------------------------
# 5. Предполётный отчёт
# ---------------------------------------------------------------------------
preflight_report() {
    log "=== Предполётный отчёт ==="
    log "Режим: $($DRY_RUN && echo DRY-RUN || echo REAL) | $([ -n "$SINCE" ] && echo "ИНКРЕМЕНТ с ${SINCE}" || echo 'ПОЛНАЯ ПЕРЕЗАГРУЗКА')"
    log "Источник: ${MSSQL_HOST}:${MSSQL_PORT}/${MSSQL_DB} :: ${MSSQL_TABLE}"
    log "STG: ${STG_TABLE} (полная перезапись)"
    log "RAW: ${RAW_TABLE} (ON CONFLICT DO NOTHING)"

    # Текущее состояние STG
    local stg_rows
    stg_rows="$(psql_at "SELECT count(*) FROM ${STG_TABLE};")"
    log "Строк в STG сейчас: ${stg_rows:-?}"

    # Текущее состояние RAW
    local raw_rows
    raw_rows="$(psql_at "SELECT count(*) FROM ${RAW_TABLE};")"
    log "Строк в RAW сейчас: ${raw_rows:-?}"

    # Партиции RAW
    local parts
    parts="$(psql_at "
        SELECT string_agg(c.relname, ', ' ORDER BY c.relname)
        FROM pg_inherits i
        JOIN pg_class c ON c.oid = i.inhrelid
        JOIN pg_class p ON p.oid = i.inhparent
        JOIN pg_namespace n ON n.oid = p.relnamespace
        WHERE p.relname = '${TABLE_NAME}' AND n.nspname = '${RAW_SCHEMA}';")"
    log "Партиции RAW: ${parts:-нет}"
}

# ---------------------------------------------------------------------------
# 6. Извлечение из MS SQL Server в STG (Extract + Load в STG)
# ---------------------------------------------------------------------------
extract_to_stg() {
    log "--- Фаза 1: MS SQL Server → STG ---"

    TMP_DIR="$(mktemp -d /tmp/stg_to_raw.XXXXXX)"
    local csv_file="${TMP_DIR}/${TABLE_NAME}.csv"

    # 6.1. Вызов Python-экстрактора (см. раздел «Компоненты»)
    local extract_args=(
        --host "$MSSQL_HOST" --port "$MSSQL_PORT"
        --database "$MSSQL_DB" --user "$MSSQL_USER"
        --password "$MSSQL_PASSWORD"
        --table "$MSSQL_TABLE"
        --output "$csv_file"
    )
    [[ -n "$SINCE" ]] && extract_args+=( --since "$SINCE" )

    if $DRY_RUN; then
        log "[DRY-RUN] python3 ${PYTHON_EXTRACTOR} ${extract_args[*]}"
        return 0
    fi

    log "Извлечение ${MSSQL_TABLE} → ${csv_file}."
    python3 "$PYTHON_EXTRACTOR" "${extract_args[@]}" \
        || die "Python-экстрактор завершился с ошибкой."

    [[ -s "$csv_file" ]] || die "CSV-файл пуст или не создан: ${csv_file}."
    local lines
    lines="$(wc -l < "$csv_file")"
    log "Извлечено строк (с заголовком): ${lines}."

    # 6.2. Загрузка CSV в STG (полная перезапись в одной транзакции)
    log "Загрузка CSV → ${STG_TABLE} (TRUNCATE + COPY)."
    run_sql "
        SET LOCAL lock_timeout = '${LOCK_TIMEOUT}';
        TRUNCATE TABLE ${STG_TABLE};
    " || die "TRUNCATE ${STG_TABLE} не удался."

    # \copy — клиентская команда psql, выполняется в рамках сессии
    if ! sudo -u "$PG_USER" -- psql -X -q -d "$PG_DB" \
            -v ON_ERROR_STOP=1 \
            -c "\copy ${STG_TABLE} (id, dt, json_data, col4, col5) FROM '${csv_file}' WITH (FORMAT csv, HEADER true)" \
            >/dev/null; then
        die "\\copy в ${STG_TABLE} не удался."
    fi
    log "STG перезагружен."

    local stg_rows
    stg_rows="$(psql_at "SELECT count(*) FROM ${STG_TABLE};")"
    log "Строк в STG после загрузки: ${stg_rows}."
}

# ---------------------------------------------------------------------------
# 7. Автосоздание партиций RAW для диапазона дат из STG
# ---------------------------------------------------------------------------
ensure_partitions_for_range() {
    log "--- Фаза 2: проверка/создание партиций RAW ---"

    # 6.1. Определяем минимальную и максимальную дату в STG
    local min_dt max_dt
    min_dt="$(psql_at "SELECT min(dt)::date FROM ${STG_TABLE};")"
    max_dt="$(psql_at "SELECT max(dt)::date FROM ${STG_TABLE};")"
    [[ -n "$min_dt" && -n "$max_dt" ]] || { log "STG пуст — партиции не требуются."; return 0; }
    log "Диапазон dt в STG: ${min_dt} … ${max_dt}."

    # 6.2. Перебираем кварталы от min_dt до max_dt
    local cur_year cur_q
    cur_year="${min_dt:0:4}"
    cur_q=$(( (10#${min_dt:5:2} - 1) / 3 + 1 ))

    local end_year="${max_dt:0:4}"
    local end_q=$(( (10#${max_dt:5:2} - 1) / 3 + 1 ))

    while (( 10#$cur_year < 10#$end_year )) || { (( 10#$cur_year == 10#$end_year )) && (( cur_q <= end_q )); }; do
        local qname="q${cur_q}_${cur_year}"
        local tsname="ts_raw_${qname}"

        # Проверяем, существует ли tablespace
        local ts_exists
        ts_exists="$(psql_at "SELECT 1 FROM pg_tablespace WHERE spcname = '${tsname}';")"

        if [[ "$ts_exists" != "1" ]]; then
            log "Tablespace ${tsname} отсутствует — вызываем create_new_partition.sh ${cur_year} ${cur_q}."
            if [[ -x "$CREATE_PARTITION_SCRIPT" ]]; then
                if $DRY_RUN; then
                    log "[DRY-RUN] ${CREATE_PARTITION_SCRIPT} ${cur_year} ${cur_q}"
                else
                    "$CREATE_PARTITION_SCRIPT" "${cur_year}" "${cur_q}" \
                        || die "create_new_partition.sh ${cur_year} ${cur_q} не удался."
                fi
            else
                die "Скрипт ${CREATE_PARTITION_SCRIPT} не найден. Создайте партицию ${qname} вручную."
            fi
        else
            log "Tablespace ${tsname} уже существует."
        fi

        # Инкремент квартала
        if (( cur_q == 4 )); then
            cur_q=1
            cur_year=$(( 10#$cur_year + 1 ))
        else
            cur_q=$(( cur_q + 1 ))
        fi
    done

    log "Партиции для диапазона ${min_dt}…${max_dt} готовы."
}

# ---------------------------------------------------------------------------
# 8. Загрузка STG → RAW (идемпотентно, ON CONFLICT DO NOTHING)
# ---------------------------------------------------------------------------
load_stg_to_raw() {
    log "--- Фаза 3: STG → RAW ---"

    local stg_rows
    stg_rows="$(psql_at "SELECT count(*) FROM ${STG_TABLE};")"
    if [[ "${stg_rows:-0}" == "0" ]]; then
        log "STG пуст — загрузка в RAW не требуется."
        return 0
    fi

    # 8.1. Загрузка с защитой от дублей
    #     Предполагается, что id — уникальный ключ в RAW.
    #     Если id не уникален — используйте ON CONFLICT (id, dt) или MERGE.
    run_sql "
        SET LOCAL lock_timeout = '${LOCK_TIMEOUT}';
        SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}';
        INSERT INTO ${RAW_TABLE} (id, dt, json_data, col4, col5)
        SELECT id, dt, json_data, col4, col5
        FROM ${STG_TABLE}
        ON CONFLICT (id) DO NOTHING;
    " || die "INSERT INTO ${RAW_TABLE} не удался."
    log "Загрузка STG → RAW завершена."

    # 8.2. Верификация: сколько строк реально добавлено
    local raw_rows
    raw_rows="$(psql_at "SELECT count(*) FROM ${RAW_TABLE};")"
    log "Строк в RAW после загрузки: ${raw_rows}."
    log "Прирост: $(( ${raw_rows:-0} - ${PRE_RAW_ROWS:-0} )) строк (мог быть 0 при повторном запуске)."
}

# ---------------------------------------------------------------------------
# 9. Гранты для прикладной роли
# ---------------------------------------------------------------------------
apply_grants() {
    run_sql "
        GRANT USAGE ON SCHEMA ${STG_SCHEMA} TO ${APP_USER};
        GRANT SELECT, INSERT, UPDATE, DELETE ON ${STG_TABLE} TO ${APP_USER};
        GRANT SELECT, INSERT ON ${RAW_TABLE} TO ${APP_USER};
    " || log "ВНИМАНИЕ: гранты для ${APP_USER} не применены."
    log "Гранты для ${APP_USER} применены к STG и RAW."
}

# ---------------------------------------------------------------------------
# 10. Main
# ---------------------------------------------------------------------------
main() {
    require_root
    acquire_lock
    log "=== Начало stg_to_raw.sh ==="

    check_dependencies
    PRE_RAW_ROWS="$(psql_at "SELECT count(*) FROM ${RAW_TABLE};")"
    preflight_report

    extract_to_stg
    ensure_partitions_for_range
    load_stg_to_raw
    apply_grants

    if $DRY_RUN; then
        log "=== DRY-RUN завершён. Реальных изменений не производилось. ==="
    else
        log "=== ETL MS SQL → STG → RAW завершён успешно. ==="
    fi
}

main "$@"

Скрипт файла stg_to_raw.sh для чтения данных из слоя STG в слой RAW¶

Файл stg_to_raw.sh — предназначен для автоматизации чтения данных из слоя STG в квартальные партиции слоя RAW для их дальнейшего хранения.

#!/usr/bin/env bash
# ============================================================================
# stg_to_raw.sh — ETL: STG → RAW (Data Lakehouse MVP)
#
# Назначение:
#   * STG — «песочница», уже наполненная данными из MS SQL Server
#     (отдельным скриптом extract_to_stg.sh).
#   * RAW — долгосрочное неизменяемое хранилище, партиционированное по
#     кварталам. Данные только добавляются (append-only).
#
# Принципы (те же, что в create_raw.sh / drop_raw.sh / archive / restore):
#   1. Идемпотентность: повторный запуск не создаёт дублей в RAW
#      (INSERT ... ON CONFLICT (id) DO NOTHING).
#   2. Атомарность: загрузка в RAW выполняется в одной транзакции.
#   3. Наблюдаемость: предполётный отчёт + верификация STG ↔ RAW.
#   4. Готовность к CI/CD: --dry-run, вынесенная конфигурация, flock.
#   5. Non-disruptive: PostgreSQL не останавливается.
#
# Зависимости:
#   * create_new_partition.sh в PATH или /opt/dlh/scripts/ (для автосоздания
#     партиций, если данные выходят за пределы существующих).
#
# Использование:
#   sudo ./stg_to_raw.sh                    # загрузить всё содержимое STG в RAW
#   sudo ./stg_to_raw.sh --since=2026-07-01 # только строки с dt >= указанной даты
#   sudo ./stg_to_raw.sh --dry-run          # показать план, ничего не менять
# ============================================================================

set -Eeuo pipefail
trap 'on_error $? $LINENO "${BASH_COMMAND}"' ERR
trap 'cleanup' EXIT

on_error() {
    log "КРИТИЧЕСКАЯ ОШИБКА: код=$1, строка=$2, команда='$3'"
    exit "$1"
}

cleanup() {
    log "Завершение stg_to_raw.sh."
}

# ---------------------------------------------------------------------------
# 1. Конфигурация
# ---------------------------------------------------------------------------
CONFIG_FILE="${CONFIG_FILE:-/etc/dlh/raw.env}"
[[ -f "$CONFIG_FILE" ]] && source "$CONFIG_FILE"

PG_VERSION="${PG_VERSION:-17}"
PG_SERVICE="${PG_SERVICE:-postgresql-${PG_VERSION}}"
PG_DB="${PG_DB:-dlh_db}"
PG_USER="${PG_USER:-postgres}"

RAW_SCHEMA="${RAW_SCHEMA:-raw}"
STG_SCHEMA="${STG_SCHEMA:-stg}"
TABLE_NAME="${TABLE_NAME:-table_1}"
STG_TABLE="${STG_SCHEMA}.${TABLE_NAME}"
RAW_TABLE="${RAW_SCHEMA}.${TABLE_NAME}"

# Роли (least privilege)
APP_USER="${APP_USER:-dlh_user}"

# Вспомогательные пути
CREATE_PARTITION_SCRIPT="${CREATE_PARTITION_SCRIPT:-/opt/dlh/scripts/create_new_partition.sh}"

LOG_DIR="${LOG_DIR:-/var/log/dlh/scripts}"
LOG_FILE="${LOG_FILE:-${LOG_DIR}/stg_to_raw_$(date +%Y%m%d_%H%M%S).log}"

# Таймауты
LOCK_TIMEOUT="${LOCK_TIMEOUT:-10s}"
STATEMENT_TIMEOUT="${STATEMENT_TIMEOUT:-300s}"

# ---------------------------------------------------------------------------
# 2. Аргументы
# ---------------------------------------------------------------------------
SINCE=""
DRY_RUN=false

usage() {
    cat <<EOF
Использование: $0 [опции]

Опции:
  --since=YYYY-MM-DD    Загрузить только строки с dt >= указанной даты.
                        Без опции — полная загрузка STG в RAW (append-only).
  --dry-run             Показать план, ничего не выполнять.
  -h, --help            Эта справка.

Конфигурация: $CONFIG_FILE (или CONFIG_FILE=... в окружении).
EOF
}

while [[ $# -gt 0 ]]; do
    case "$1" in
        --since=*)   SINCE="${1#*=}" ;;
        --dry-run)   DRY_RUN=true ;;
        -h|--help)   usage; exit 0 ;;
        *)           echo "Неизвестная опция: $1" >&2; usage; exit 2 ;;
    esac
    shift
done

if [[ -n "$SINCE" ]] && ! [[ "$SINCE" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]]; then
    echo "ОШИБКА: --since должен быть в формате YYYY-MM-DD: ${SINCE}" >&2
    exit 2
fi

# ---------------------------------------------------------------------------
# 3. Базовые функции
# ---------------------------------------------------------------------------
mkdir -p "$LOG_DIR"

log() {
    local msg; msg="$(date '+%Y-%m-%d %H:%M:%S') - $1"
    echo "$msg" | tee -a "$LOG_FILE" >&2
}

die() { log "ФАТАЛЬНАЯ ОШИБКА: $1"; exit 1; }

require_root() { [[ $EUID -eq 0 ]] || die "Требуется root (sudo)."; }

acquire_lock() {
    exec 9>"/var/lock/stg_to_raw.lock"
    flock -n 9 || die "Другой экземпляр stg_to_raw.sh уже выполняется."
}

run_cmd() {
    local desc="$1"; shift
    if $DRY_RUN; then log "[DRY-RUN] ${desc}"; return 0; fi
    log "${desc}"; "$@"
}

run_sql() {
    local sql="$1"; shift
    if $DRY_RUN; then
        log "[DRY-RUN] SQL: $(printf '%s' "$sql" | tr '\n' ' ' | head -c 400)"
        return 0
    fi
    log "SQL (${#sql} байт)."
    printf '%s\n' "$sql" |
        sudo -u "$PG_USER" -- psql -X -q -d "$PG_DB" \
            -v ON_ERROR_STOP=1 -1 "$@" -f - >/dev/null \
        || { log "Ошибка SQL."; return 1; }
}

psql_at() {
    sudo -u "$PG_USER" -- psql -X -At -d "$PG_DB" -v ON_ERROR_STOP=1 -c "$1" 2>/dev/null || true
}

# ---------------------------------------------------------------------------
# 4. Проверка зависимостей
# ---------------------------------------------------------------------------
check_dependencies() {
    log "--- Проверка зависимостей ---"

    # PostgreSQL
    if ! systemctl is-active --quiet "$PG_SERVICE"; then
        run_cmd "Запуск ${PG_SERVICE}" systemctl start "$PG_SERVICE" \
            || die "Не удалось запустить ${PG_SERVICE}."
    fi
    for _ in {1..30}; do
        sudo -u "$PG_USER" -- pg_isready -q && break
        sleep 1
    done
    sudo -u "$PG_USER" -- pg_isready -q || die "PostgreSQL не готов."

    # STG-таблица должна существовать и содержать данные
    local stg_exists
    stg_exists="$(psql_at "SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='${STG_SCHEMA}' AND c.relname='${TABLE_NAME}';")"
    [[ "$stg_exists" == "1" ]] || die "STG-таблица ${STG_TABLE} не найдена."

    # RAW-таблица должна существовать и быть партиционированной
    local raw_exists
    raw_exists="$(psql_at "SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='${RAW_SCHEMA}' AND c.relname='${TABLE_NAME}' AND c.relkind='p';")"
    [[ "$raw_exists" == "1" ]] || die "RAW-таблица ${RAW_TABLE} не найдена или не является партиционированной."

    log "Все зависимости в порядке."
}

# ---------------------------------------------------------------------------
# 5. Предполётный отчёт
# ---------------------------------------------------------------------------
preflight_report() {
    log "=== Предполётный отчёт ==="
    log "Режим: $($DRY_RUN && echo DRY-RUN || echo REAL) | $([ -n "$SINCE" ] && echo "ИНКРЕМЕНТ с ${SINCE}" || echo 'ПОЛНАЯ ЗАГРУЗКА STG')"
    log "STG: ${STG_TABLE}"
    log "RAW: ${RAW_TABLE}"

    local stg_rows raw_rows
    stg_rows="$(psql_at "SELECT count(*) FROM ${STG_TABLE};")"
    raw_rows="$(psql_at "SELECT count(*) FROM ${RAW_TABLE};")"
    log "Строк в STG: ${stg_rows:-?} | Строк в RAW: ${raw_rows:-?}"

    local parts
    parts="$(psql_at "
        SELECT string_agg(c.relname, ', ' ORDER BY c.relname)
        FROM pg_inherits i
        JOIN pg_class c ON c.oid = i.inhrelid
        JOIN pg_class p ON p.oid = i.inhparent
        JOIN pg_namespace n ON n.oid = p.relnamespace
        WHERE p.relname = '${TABLE_NAME}' AND n.nspname = '${RAW_SCHEMA}';")"
    log "Партиции RAW: ${parts:-нет}"
}

# ---------------------------------------------------------------------------
# 6. Автосоздание партиций RAW для диапазона дат из STG
# ---------------------------------------------------------------------------
ensure_partitions_for_range() {
    log "--- Фаза 1: проверка/создание партиций RAW ---"

    local min_dt max_dt
    min_dt="$(psql_at "SELECT min(dt)::date FROM ${STG_TABLE}${SINCE:+ WHERE dt >= '${SINCE}'};")"
    max_dt="$(psql_at "SELECT max(dt)::date FROM ${STG_TABLE}${SINCE:+ WHERE dt >= '${SINCE}'};")"
    [[ -n "$min_dt" && -n "$max_dt" ]] || { log "STG пуст — партиции не требуются."; return 0; }
    log "Диапазон dt в STG: ${min_dt} … ${max_dt}."

    local cur_year cur_q
    cur_year="${min_dt:0:4}"
    cur_q=$(( (10#${min_dt:5:2} - 1) / 3 + 1 ))
    local end_year="${max_dt:0:4}"
    local end_q=$(( (10#${max_dt:5:2} - 1) / 3 + 1 ))

    while (( 10#$cur_year < 10#$end_year )) || { (( 10#$cur_year == 10#$end_year )) && (( cur_q <= end_q )); }; do
        local qname="q${cur_q}_${cur_year}"
        local tsname="ts_raw_${qname}"

        if [[ "$(psql_at "SELECT 1 FROM pg_tablespace WHERE spcname = '${tsname}';")" != "1" ]]; then
            log "Tablespace ${tsname} отсутствует — вызываем create_new_partition.sh ${cur_year} ${cur_q}."
            if [[ -x "$CREATE_PARTITION_SCRIPT" ]]; then
                if $DRY_RUN; then
                    log "[DRY-RUN] ${CREATE_PARTITION_SCRIPT} ${cur_year} ${cur_q}"
                else
                    "$CREATE_PARTITION_SCRIPT" "${cur_year}" "${cur_q}" \
                        || die "create_new_partition.sh ${cur_year} ${cur_q} не удался."
                fi
            else
                die "Скрипт ${CREATE_PARTITION_SCRIPT} не найден. Создайте партицию ${qname} вручную."
            fi
        else
            log "Tablespace ${tsname} уже существует."
        fi

        if (( cur_q == 4 )); then
            cur_q=1
            cur_year=$(( 10#$cur_year + 1 ))
        else
            cur_q=$(( cur_q + 1 ))
        fi
    done
    log "Партиции для диапазона ${min_dt}…${max_dt} готовы."
}

# ---------------------------------------------------------------------------
# 7. Загрузка STG → RAW (идемпотентно, ON CONFLICT DO NOTHING)
# ---------------------------------------------------------------------------
load_stg_to_raw() {
    log "--- Фаза 2: STG → RAW ---"

    local where_clause=""
    [[ -n "$SINCE" ]] && where_clause="WHERE dt >= '${SINCE}'"

    local src_rows
    src_rows="$(psql_at "SELECT count(*) FROM ${STG_TABLE} ${where_clause};")"
    if [[ "${src_rows:-0}" == "0" ]]; then
        log "STG пуст (или нет строк с dt >= ${SINCE}) — загрузка не требуется."
        return 0
    fi
    log "К загрузке из STG: ${src_rows} строк."

    # Идемпотентная вставка: ON CONFLICT (id) DO NOTHING.
    # Если id не уникален — используйте ON CONFLICT (id, dt) или MERGE.
    run_sql "
        SET LOCAL lock_timeout = '${LOCK_TIMEOUT}';
        SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}';
        INSERT INTO ${RAW_TABLE} (id, dt, json_data, col4, col5)
        SELECT id, dt, json_data, col4, col5
        FROM ${STG_TABLE}
        ${where_clause}
        ON CONFLICT (id) DO NOTHING;
    " || die "INSERT INTO ${RAW_TABLE} не удался."
    log "Загрузка STG → RAW завершена."

    local raw_rows
    raw_rows="$(psql_at "SELECT count(*) FROM ${RAW_TABLE};")"
    log "Строк в RAW после загрузки: ${raw_rows}."
    log "Прирост: $(( ${raw_rows:-0} - ${PRE_RAW_ROWS:-0} )) строк (0 — если повторный запуск)."
}

# ---------------------------------------------------------------------------
# 8. Гранты для прикладной роли
# ---------------------------------------------------------------------------
apply_grants() {
    run_sql "
        GRANT USAGE ON SCHEMA ${STG_SCHEMA} TO ${APP_USER};
        GRANT SELECT, INSERT, UPDATE, DELETE ON ${STG_TABLE} TO ${APP_USER};
        GRANT SELECT, INSERT ON ${RAW_TABLE} TO ${APP_USER};
    " || log "ВНИМАНИЕ: гранты для ${APP_USER} не применены."
    log "Гранты для ${APP_USER} применены к STG и RAW."
}

# ---------------------------------------------------------------------------
# 9. Main
# ---------------------------------------------------------------------------
main() {
    require_root
    acquire_lock
    log "=== Начало stg_to_raw.sh (STG → RAW) ==="

    check_dependencies
    PRE_RAW_ROWS="$(psql_at "SELECT count(*) FROM ${RAW_TABLE};")"
    preflight_report

    ensure_partitions_for_range
    load_stg_to_raw
    apply_grants

    if $DRY_RUN; then
        log "=== DRY-RUN завершён. Реальных изменений не производилось. ==="
    else
        log "=== ETL STG → RAW завершён успешно. ==="
    fi
}

main "$@"

AI-Ready платформа Bash ELT PostgreSQL RAW STG гибридное хранилище данных корпоративная информационная система практика

Предыдущая статьяАвтоматизация управления жизненным циклом партиций слоя RAW гибридного хранилища данных

Рубрики

Метки

abc abcd AI-Ready платформа Bash Data Lakehouse ELT excel ms sql n8n PostgreSQL PowerShell Python RAG RAW sql STG tessa VBA xyz ИИ-агент анализ виртуальный помощник гибридное хранилище данных данные знания информационная система информация искусственный интеллект кластерный анализ компетенции корпоративная информационная система математика модель предоставления прав мудрость о проекте оптимизация практика программное обеспечение ролевая модель сеть; теория теория вероятностей тесса юмор языки программирования

Политика конфиденциальности

Продолжая использовать данный сайт вы подтверждаете свое согласие с условиями его политики конфиденциальности. Подробнее…




Администрация и владельцы данного информационного ресурса не несут ответственности за возможные последствия, связанные с использованием информации, размещенной на нем.


Все права защищены. При копировании материалов сайта обязательно указывать ссылку на © Microsegment.ru (2020-2026)