0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@naritomo08(nari tomo)

別Hadoop Hive ParquetデータをElasticsearch Data Streamへ追加投入する

0
Last updated at Posted at 2026-06-14

別Hadoop Hive ParquetデータをElasticsearch Data Streamへ追加投入する

概要

既存の Iceberg → Elasticsearch Data Stream 構成に加え、別Hadoop上の Hive Parquet テーブルを Elasticsearch Data Stream へ投入します。

前提

  • Elasticsearch: http://elastic1:9200
  • HiveServer2: jdbc:hive2://hdptest:10000/default
  • syslog: logs.sysloglocal_curated
  • authlog: authlogs.authlog_curated

投入先:

  • logs-syslog-hive-YYYY.MM.DD
  • logs-authlog-hive-YYYY.MM.DD

Index Template作成

登録

curl -X PUT "http://elastic1:9200/_index_template/logs_syslog_hive_template" \
  -H "Content-Type: application/json" \
  -d '{
    "index_patterns": ["logs-syslog-hive-*"],
    "priority": 1500,
    "data_stream": {},
    "template": {
      "lifecycle": {
        "data_retention": "14d"
      },
      "mappings": {
        "properties": {
          "@timestamp": {
            "type": "date"
          },
          "dt": {
            "type": "date",
            "format": "strict_date||yyyy-MM-dd"
          },
          "hr": {
            "type": "integer"
          },
          "host": {
            "type": "keyword"
          },
          "program": {
            "type": "keyword"
          },
          "severity": {
            "type": "integer"
          },
          "msg": {
            "type": "text"
          }
        }
      }
    }
  }'

curl -X PUT "http://elastic1:9200/_index_template/logs_authlog_hive_template" \
  -H "Content-Type: application/json" \
  -d '{
    "index_patterns": ["logs-authlog-hive-*"],
    "priority": 1500,
    "data_stream": {},
    "template": {
      "lifecycle": {
        "data_retention": "14d"
      },
      "mappings": {
        "properties": {
          "@timestamp": {
            "type": "date"
          },
          "dt": {
            "type": "date",
            "format": "strict_date||yyyy-MM-dd"
          },
          "hr": {
            "type": "integer"
          },
          "host": {
            "type": "keyword"
          },
          "program": {
            "type": "keyword"
          },
          "severity": {
            "type": "integer"
          },
          "msg": {
            "type": "text"
          }
        }
      }
    }
  }'

確認

curl -s "http://elastic1:9200/_index_template/logs_syslog_hive_template?pretty"

curl -s "http://elastic1:9200/_index_template/logs_authlog_hive_template?pretty"

ディレクトリ作成

sudo mkdir -p /opt/elastic/bin
sudo mkdir -p /var/log/elastic

Hive → Elasticsearch投入シェル

配置先:

ope1で配置すること。

/opt/elastic/bin/export_hive_log_to_es.sh

sudo tee /opt/elastic/bin/export_hive_log_to_es.sh >/dev/null <<'EOF'
#!/usr/bin/env bash
set -euo pipefail

DATASET="${1:?usage: $0 syslog|authlog [YYYY-MM-DD]}"
TARGET_DT="${2:-$(date -d 'yesterday' +%F)}"

ES_URL="${ES_URL:-http://elastic1:9200}"
BEELINE_URL="${BEELINE_URL:-jdbc:hive2://hdptest:10000/default}"
TMP_DIR="${TMP_DIR:-/tmp/hive_to_es}"
CHUNK_LINES="${CHUNK_LINES:-20000}"

SYS_TBL="${SYS_TBL:-logs.syslog_curated}"
AUTH_TBL="${AUTH_TBL:-authlogs.authlog_curated}"

mkdir -p "${TMP_DIR}"

if ! [[ "${TARGET_DT}" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]]; then
  echo "[ERROR] TARGET_DT must be YYYY-MM-DD. input=${TARGET_DT}"
  exit 1
fi

case "${DATASET}" in
  syslog)
    HIVE_TABLE="${SYS_TBL}"
    DS_NAME="logs-syslog-hive-${TARGET_DT//-/.}"
    ;;
  authlog)
    HIVE_TABLE="${AUTH_TBL}"
    DS_NAME="logs-authlog-hive-${TARGET_DT//-/.}"
    ;;
  *)
    echo "[ERROR] DATASET must be syslog or authlog"
    exit 1
    ;;
esac

CSV_FILE="${TMP_DIR}/${DATASET}.csv"
BULK_FILE="${TMP_DIR}/${DATASET}.ndjson"
RESULT_FILE="${TMP_DIR}/${DATASET}.bulk.result.json"
DELETE_RESULT="${TMP_DIR}/delete_${DS_NAME}.json"
CHUNK_DIR="${TMP_DIR}/${DATASET}_${TARGET_DT}_chunks"

echo "[INFO] DATASET=${DATASET}"
echo "[INFO] TARGET_DT=${TARGET_DT}"
echo "[INFO] HIVE_TABLE=${HIVE_TABLE}"
echo "[INFO] DATA_STREAM=${DS_NAME}"
echo "[INFO] ES_URL=${ES_URL}"
echo "[INFO] BEELINE_URL=${BEELINE_URL}"
echo "[INFO] CHUNK_LINES=${CHUNK_LINES}"

rm -f "${CSV_FILE}" "${BULK_FILE}" "${RESULT_FILE}"
rm -rf "${CHUNK_DIR}"

echo "[INFO] export hive table to DSV"

beeline \
  -u "${BEELINE_URL}" \
  --silent=true \
  --showHeader=false \
  --outputformat=dsv \
  --delimiterForDSV=$'\001' \
  -e "
SELECT
  host,
  ts_text as ts,
  severity,
  program,
  msg,
  dt
FROM ${HIVE_TABLE}
WHERE dt = '${TARGET_DT}'
" > "${CSV_FILE}"

echo "[INFO] CSV_FILE=${CSV_FILE}"
echo "[INFO] create bulk ndjson"

python3 - <<PY
import csv
import json
import re
import sys

csv_file = "${CSV_FILE}"
bulk_file = "${BULK_FILE}"
ds_name = "${DS_NAME}"
dataset = "${DATASET}"

def normalize(v):
    if v is None:
        return None
    v = v.strip()
    if v == "" or v.upper() == "NULL":
        return None
    return v

count = 0
skip_count = 0

with open(csv_file, newline="", encoding="utf-8") as f, open(bulk_file, "w", encoding="utf-8") as out:
    reader = csv.reader(f, delimiter="\001")

    for row in reader:
        if len(row) < 6:
            skip_count += 1
            continue

        host, ts, severity, program, msg, dt = row[:6]

        host = normalize(host)
        ts = normalize(ts)
        severity = normalize(severity)
        program = normalize(program)
        msg = normalize(msg)
        dt = normalize(dt)

        if ts is None or dt is None:
            skip_count += 1
            continue

        if not re.match(r"^\d{4}-\d{2}-\d{2}$", dt):
            skip_count += 1
            continue

        try:
            severity_value = None if severity is None else int(severity)
        except ValueError:
            severity_value = None

        doc = {
            "@timestamp": ts.replace(" ", "T") + "+09:00",
            "host": host,
            "severity": severity_value,
            "program": program,
            "msg": msg,
            "dt": dt,
            "source": "hive_parquet",
            "dataset": dataset
        }

        action = {
            "create": {
                "_index": ds_name
            }
        }

        out.write(json.dumps(action, ensure_ascii=False) + "\\n")
        out.write(json.dumps(doc, ensure_ascii=False) + "\\n")
        count += 1

print(f"[INFO] docs={count}", file=sys.stderr)
print(f"[INFO] skipped_rows={skip_count}", file=sys.stderr)
PY

if [ ! -s "${BULK_FILE}" ]; then
  echo "[WARN] BULK_FILE is empty. no data to send: ${BULK_FILE}"
  exit 0
fi

echo "[INFO] delete existing data stream if exists: ${DS_NAME}"

http_code=$(curl -s -o "${DELETE_RESULT}" -w "%{http_code}" \
  -X DELETE "${ES_URL}/_data_stream/${DS_NAME}" || true)

case "${http_code}" in
  200)
    echo "[INFO] deleted existing data stream: ${DS_NAME}"
    ;;
  404)
    echo "[INFO] data stream does not exist. continue: ${DS_NAME}"
    ;;
  *)
    echo "[ERROR] failed to delete data stream. http_code=${http_code}"
    cat "${DELETE_RESULT}" || true
    exit 1
    ;;
esac

echo "[INFO] split bulk file"

mkdir -p "${CHUNK_DIR}"

split -l "${CHUNK_LINES}" -d -a 5 "${BULK_FILE}" "${CHUNK_DIR}/chunk_"

echo "[INFO] bulk import to Elasticsearch by chunks"

chunk_no=0

for chunk in "${CHUNK_DIR}"/chunk_*
do
  chunk_no=$((chunk_no + 1))
  chunk_result="${chunk}.result.json"

  echo "[INFO] bulk chunk ${chunk_no}: ${chunk}"

  curl -sS \
    -H "Content-Type: application/x-ndjson" \
    -X POST "${ES_URL}/_bulk" \
    --data-binary @"${chunk}" \
    > "${chunk_result}"

  if [ ! -s "${chunk_result}" ]; then
    echo "[ERROR] empty response from Elasticsearch. chunk=${chunk}"
    exit 1
  fi

  ERRORS=$(jq -r '.errors' "${chunk_result}")

  echo "[INFO] chunk ${chunk_no} errors=${ERRORS}"

  if [ "${ERRORS}" != "false" ]; then
    echo "[ERROR] bulk insert failed. chunk=${chunk}"
    jq '.items[] | select(.create.error != null) | .create.error' "${chunk_result}" | head -20
    exit 1
  fi
done

echo "[INFO] all bulk chunks completed"

echo "[INFO] refresh"

curl -sS \
  -X POST \
  "${ES_URL}/${DS_NAME}/_refresh" >/dev/null

echo "[INFO] count after import"

curl -s \
  "${ES_URL}/${DS_NAME}/_count?pretty"

echo "[INFO] done"
EOF

sudo chmod 755 /opt/elastic/bin/export_hive_log_to_es.sh

実行

/opt/elastic/bin/export_hive_log_to_es.sh syslog 2026-06-14

/opt/elastic/bin/export_hive_log_to_es.sh authlog 2026-06-14

日次ラッパ

/opt/elastic/bin/export_hive_logs_daily.sh

sudo tee /opt/elastic/bin/export_hive_logs_daily.sh >/dev/null <<'EOF'
#!/usr/bin/env bash
set -euo pipefail

TARGET_DT="${1:-$(date -d 'yesterday' +%F)}"

/opt/elastic/bin/export_hive_log_to_es.sh syslog "${TARGET_DT}"
/opt/elastic/bin/export_hive_log_to_es.sh authlog "${TARGET_DT}"
EOF

sudo chmod +x /opt/elastic/bin/export_hive_logs_daily.sh

実行

/opt/elastic/bin/export_hive_logs_daily.sh

Data Stream Lifecycle による保持期間設定

本記事では Elasticsearch の Data Stream Lifecycle (DLM) を利用し、
古いデータを自動削除します。

Data Stream Lifecycle を利用することで、

  • cronによる削除処理が不要
  • 保持期間をElasticsearch側で管理
  • 運用シェルを削減できる

ようになります。

インデックス設定ですでに対応しています。

確認

curl \
http://elastic1:9200/_data_stream/logs-syslog-hive-*?human

curl \
http://elastic1:9200/_data_stream/logs-authlog-hive-*?human

出力例

...
"lifecycle" : {
  "enabled" : true,
  "data_retention" : "14d"
}

設定後は14日を経過したデータストリームが
Elasticsearchによって自動的に削除されます。

cron

30 1 * * * /opt/elastic/bin/export_hive_logs_daily.sh

確認

curl -s http://elastic1:9200/_data_stream | jq

curl -s \
"http://elastic1:9200/logs-syslog-hive-2026.06.14/_count?pretty"

curl -s \
"http://elastic1:9200/logs-authlog-hive-2026.06.14/_count?pretty"

Kibana

Data View:

logs-syslog-hive-*
logs-authlog-hive-*

または

logs-*

検索アプリ

検索対象が

logs-*

なら Hive由来も検索対象になります。

識別用フィールド:

{
  "source":"hive_parquet",
  "dataset":"syslog"
}

{
  "source":"hive_parquet",
  "dataset":"authlog"
}
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?