数据转换逻辑设计

TSV conversion logic for logs

Scope

• Input: arbitrary application logs in either text or JSON-lines.
• Output: tab-separated values (TSV) with a defined header. One record per log event.
• Constraints: streaming-friendly, robust to missing fields, tabs/newlines in values, mixed formats, and occasional multi-line entries.

1. Target TSV schema Use a fixed schema to maintain streaming compatibility. Recommended minimal columns:

• ts_iso: normalized timestamp in ISO 8601 (UTC if tz provided; otherwise keep naive).
• level: normalized log level (TRACE/DEBUG/INFO/WARN/ERROR/FATAL).
• service: service or application name (if discoverable).
• host: hostname or source (if discoverable).
• logger: logger name/category (if present).
• thread: thread identifier (if present).
• request_id: request or correlation id (if present).
• user_id: user identifier (if present).
• client_ip: client IP (if present).
• method: HTTP method (if present).
• path: URL path (if present).
• status: HTTP status (if present).
• latency_ms: request latency in milliseconds (if parseable).
• message: human-readable message portion.
• extras_json: JSON-encoded string of any remaining key-value pairs.
• raw: original log line for auditability.

Notes:

• If your environment requires a different set of fixed columns, adjust accordingly. Keep an extras_json catch-all for forward compatibility.

2. Detection and parsing strategy The parser operates per line and chooses one of two branches:

A) JSON log line

• Condition: line starts with “{” after leading whitespace.
• Action:
  • Parse JSON into a dict; if parsing fails, treat as text log (branch B).
  • Flatten to a single level: nested keys joined by dots (e.g., http.request.method → http.request.method) or use a predetermined mapping.
  • Map known fields into the target schema (see section 3).
  • Put unmapped fields into extras_json.

B) Text log line with optional key=value pairs

• Identify timestamp:
  • Try common patterns via regex and datetime parsing:
    • YYYY-MM-DD HH:MM:SS,mmm
    • YYYY-MM-DD HH:MM:SS
    • YYYY-MM-DDTHH:MM:SSZ
    • YYYY-MM-DDTHH:MM:SS.sssZ
    • Syslog-like: MMM dd HH:MM:SS (year unknown; if no year, keep as original string)
• Identify level token (case-insensitive): TRACE|DEBUG|INFO|WARN|WARNING|ERROR|CRITICAL|FATAL.
• Extract key=value pairs:
  • Regex: (\w[\w.-])=("([^"\]|\.)"|\S+)
    • Supports quoted values with spaces; unescape quotes and backslashes.
    • Key may include dot or dash.
• Message extraction:
  • Take the substring after [timestamp][optional logger/thread][level] up to the first key=value pair; trim.
  • If no key=value pairs, the remainder is the message.
• Map known keys to schema; place remaining pairs in extras_json.

3. Field mapping and normalization Mapping rules (examples; adapt to your environment):

• Timestamp (ts_iso):

  • For JSON logs: prefer fields like time, timestamp, ts, @timestamp.
  • For text logs: use the first recognized timestamp token.
  • Normalization: convert to ISO 8601. If timezone provided, convert to UTC (e.g., 2025-09-26T12:05:34.123Z). If no timezone, output naive ISO 8601 without Z.
  • If parsing fails, leave ts_iso empty and retain the original in extras_json.timestamp_raw.

• Level:

  • Map aliases: WARNING→WARN, CRITICAL→ERROR (or FATAL if preferred), VERBOSE→DEBUG.
  • Output uppercase canonical values.

• Service, host, logger, thread:

  • From keys: service, app, application; host, hostname; logger, category; thread, tid. If not present, leave empty.

• HTTP-related:

  • method from method, http_method, http.request.method.
  • path from path, url, uri, http.request.path.
  • status from status, http.status, response.status.
  • client_ip from client_ip, ip, remote_addr, http.client_ip.

• Request/user identifiers:

  • request_id from request_id, req_id, correlation_id, trace_id, x_request_id.
  • user_id from user_id, uid, user.id, principal.

• Latency:

  • Accept values like “123ms”, “1.5s”, “800us”, numeric milliseconds.
  • Normalize:
    • ms → value
    • s → value × 1000
    • us/µs → value / 1000
    • ns → value / 1_000_000
  • If unitless integer/float, treat as milliseconds.
  • If unparsable, leave latency_ms empty and persist original under extras_json.latency_raw.

• Message:

  • If JSON logs: use message, msg, log, event as primary; else let message be empty.
  • For text logs: the extracted message substring.

• extras_json:

  • JSON-encode remaining unmapped key-value pairs.
  • Ensure tab/newline escaping (see section 4).

• raw:

  • The original line, unmodified (except escaping tabs/newlines when outputting TSV).

4. Escaping and TSV output rules

• Separator: single tab character between fields.
• Header: emit once at the start in the specified column order.
• Value sanitation:
  • Replace literal tab with \t.
  • Replace newline with \n and carriage return with \r.
  • Remove or escape other control characters as needed.
  • For extras_json, ensure it is a compact JSON string with the same escaping rules applied to embedded tabs/newlines.
• Missing values: output as empty string.
• No quoting in TSV; rely on escaping.

5. Multi-line log handling

• If a line lacks a timestamp and level and does not look like JSON, treat it as a continuation of the previous record’s message.
• Append to previous message with a literal \n inserted.
• If strict one-line-per-record is required, disable multi-line merge and include the entire raw line as a separate row with empty fields.

6. Example

Input lines:

1. 2025-09-26 12:05:34,123 INFO request_id=abc123 user_id=42 method=GET path=/api/v1/items latency=12ms Completed request
2. {"time":"2025-09-26T12:05:35.001+00:00","level":"warn","service":"checkout","http":{"method":"POST","path":"/orders"},"latency_ms":75,"message":"Payment pending"}

Output TSV (header + rows): ts_iso level service host logger thread request_id user_id client_ip method path status latency_ms message extras_json raw 2025-09-26T12:05:34.123 INFO abc123 42 GET /api/v1/items 12 Completed request {} 2025-09-26 12:05:34,123 INFO request_id=abc123 user_id=42 method=GET path=/api/v1/items latency=12ms Completed request 2025-09-26T12:05:35.001Z WARN checkout {"http.method":"POST","http.path":"/orders"} 75 Payment pending {"http.method":"POST","http.path":"/orders"} {"time":"2025-09-26T12:05:35.001+00:00","level":"warn","service":"checkout","http":{"method":"POST","path":"/orders"},"latency_ms":75,"message":"Payment pending"}

Note: Tabs are shown conceptually above; actual output will use tab characters. Newlines/tabs inside fields must be escaped.

7. Reference implementation (Python, streaming)

This implementation handles both JSON and text logs, performs normalization, and writes TSV to stdout. It avoids non-standard libraries.

import sys
import json
import re
from datetime import datetime, timezone

# Precompiled regexes
TS_PATTERNS = [
    ("%Y-%m-%d %H:%M:%S,%f", None),
    ("%Y-%m-%d %H:%M:%S", None),
    ("%Y-%m-%dT%H:%M:%S.%f%z", "utc"),
    ("%Y-%m-%dT%H:%M:%S%z", "utc"),
    ("%Y-%m-%dT%H:%M:%S.%fZ", "z"),
    ("%Y-%m-%dT%H:%M:%SZ", "z"),
]
LEVEL_RE = re.compile(r'\b(trace|debug|info|warn|warning|error|critical|fatal)\b', re.IGNORECASE)
KV_RE = re.compile(r'(\w[\w\.\-]*)=("([^"\\]|\\.)*"|\S+)')

HEADER = [
    "ts_iso","level","service","host","logger","thread","request_id","user_id","client_ip",
    "method","path","status","latency_ms","message","extras_json","raw"
]

def normalize_level(lvl):
    if not lvl: return ""
    l = lvl.upper()
    if l == "WARNING": return "WARN"
    if l == "CRITICAL": return "ERROR"  # adjust to FATAL if preferred
    return l

def parse_ts(s):
    s = s.strip()
    for fmt, mode in TS_PATTERNS:
        try:
            if "%z" in fmt:
                dt = datetime.strptime(s, fmt)
                if mode == "utc":
                    return dt.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
                elif mode == "z":
                    # Already Zulu; ensure milliseconds where possible
                    if dt.microsecond:
                        return dt.isoformat(timespec="milliseconds").replace("+00:00", "Z")
                    return dt.replace(tzinfo=timezone.utc).isoformat().replace("+00:00","Z")
            else:
                dt = datetime.strptime(s, fmt)
                # naive timestamp
                if "%f" in fmt:
                    return dt.isoformat(timespec="milliseconds")
                else:
                    return dt.isoformat()
        except ValueError:
            continue
    return ""  # failed

def sanitize(v):
    if v is None: return ""
    s = str(v)
    return s.replace("\t","\\t").replace("\n","\\n").replace("\r","\\r")

def unquote(value):
    # Remove surrounding quotes and unescape \" and \\ inside
    if len(value) >= 2 and value[0] == '"' and value[-1] == '"':
        inner = value[1:-1]
        inner = inner.replace('\\"', '"').replace('\\\\', '\\')
        return inner
    return value

def normalize_latency(v):
    if v is None: return ""
    s = str(v).strip().lower()
    try:
        if s.endswith("ms"):
            return str(float(s[:-2]))
        if s.endswith("s"):
            return str(float(s[:-1]) * 1000.0)
        if s.endswith("us") or s.endswith("µs"):
            return str(float(s[:-2]) / 1000.0)
        if s.endswith("ns"):
            return str(float(s[:-2]) / 1_000_000.0)
        # unitless: assume milliseconds
        return str(float(s))
    except ValueError:
        return ""  # leave raw in extras_json if needed

def parse_json_line(line):
    try:
        obj = json.loads(line)
    except json.JSONDecodeError:
        return None
    flat = {}

    def flatten(prefix, val):
        if isinstance(val, dict):
            for k, v in val.items():
                nk = f"{prefix}.{k}" if prefix else k
                flatten(nk, v)
        else:
            flat[prefix] = val

    flatten("", obj)

    row = {k: "" for k in HEADER}
    row["raw"] = line.rstrip("\n")

    # Timestamp candidates
    for key in ("@timestamp","time","timestamp","ts"):
        if key in flat:
            row["ts_iso"] = parse_ts(str(flat[key]))
            if not row["ts_iso"]:
                row["extras_json"] = json.dumps({"timestamp_raw": flat[key]}, separators=(",",":"))
            break

    row["level"]   = normalize_level(flat.get("level") or flat.get("log.level") or flat.get("severity"))
    row["service"] = str(flat.get("service") or flat.get("app") or flat.get("application") or "")

    row["host"]    = str(flat.get("host") or flat.get("hostname") or "")
    row["logger"]  = str(flat.get("logger") or flat.get("log.logger") or "")
    row["thread"]  = str(flat.get("thread") or flat.get("tid") or "")
    row["request_id"] = str(flat.get("request_id") or flat.get("req_id") or flat.get("correlation_id") or flat.get("trace_id") or "")
    row["user_id"] = str(flat.get("user_id") or flat.get("user.id") or flat.get("uid") or "")
    row["client_ip"] = str(flat.get("client_ip") or flat.get("ip") or flat.get("remote_addr") or flat.get("http.client_ip") or "")

    row["method"]  = str(flat.get("method") or flat.get("http.method") or flat.get("http.request.method") or "")
    row["path"]    = str(flat.get("path") or flat.get("url") or flat.get("uri") or flat.get("http.path") or flat.get("http.request.path") or "")
    row["status"]  = str(flat.get("status") or flat.get("http.status") or flat.get("response.status") or "")

    latency = flat.get("latency_ms") or flat.get("latency") or flat.get("http.latency") or None
    row["latency_ms"] = normalize_latency(latency)

    row["message"] = str(flat.get("message") or flat.get("msg") or flat.get("log") or flat.get("event") or "")

    # Build extras from unmapped keys
    mapped = set([
        "@timestamp","time","timestamp","ts","level","log.level","severity","service","app","application",
        "host","hostname","logger","log.logger","thread","tid","request_id","req_id","correlation_id","trace_id",
        "user_id","user.id","uid","client_ip","ip","remote_addr","http.client_ip",
        "method","http.method","http.request.method","path","url","uri","http.path","http.request.path",
        "status","http.status","response.status","latency","latency_ms","http.latency","message","msg","log","event"
    ])
    extras = {k: v for k, v in flat.items() if k not in mapped}
    row["extras_json"] = json.dumps(extras, separators=(",",":")) if extras else "{}"

    return row

def parse_text_line(line):
    row = {k: "" for k in HEADER}
    row["raw"] = line.rstrip("\n")

    s = line.strip()

    # Attempt to find timestamp at start substrings
    ts_match = None
    # Heuristic: check first ~40 chars for timestamp tokens separated by space or 'T'
    head = s[:40]
    # Try direct parsing by slicing common lengths
    for token_len in (23, 19, 20, 24):  # rough candidates
        token = head[:token_len]
        ts_iso = parse_ts(token)
        if ts_iso:
            row["ts_iso"] = ts_iso
            ts_match = token
            break

    # Level
    level_m = LEVEL_RE.search(s)
    if level_m:
        row["level"] = normalize_level(level_m.group(0))

    # KV pairs
    kvs = []
    for m in KV_RE.finditer(s):
        key = m.group(1)
        val = unquote(m.group(2))
        kvs.append((key, val))

    # Message: substring after level up to first kv pair
    msg = s
    if level_m:
        start = level_m.end()
        msg = s[start:].strip()
    if kvs:
        first_kv_start = s.find(kvs[0][0] + "=")
        if first_kv_start != -1:
            msg = s[:first_kv_start].strip()
    row["message"] = msg

    # Map common keys
    def get(klist):
        for k in klist:
            for kk, vv in kvs:
                if kk == k:
                    return vv
        return None

    row["service"]    = get(["service","app","application"]) or ""
    row["host"]       = get(["host","hostname"]) or ""
    row["logger"]     = get(["logger","category"]) or ""
    row["thread"]     = get(["thread","tid"]) or ""
    row["request_id"] = get(["request_id","req_id","correlation_id","trace_id","x_request_id"]) or ""
    row["user_id"]    = get(["user_id","uid","user.id","principal"]) or ""
    row["client_ip"]  = get(["client_ip","ip","remote_addr"]) or ""
    row["method"]     = get(["method","http_method"]) or ""
    row["path"]       = get(["path","url","uri"]) or ""
    row["status"]     = get(["status","http_status"]) or ""
    row["latency_ms"] = normalize_latency(get(["latency","latency_ms","duration","elapsed"]))

    # Build extras
    mapped_keys = set([
        "service","app","application","host","hostname","logger","category","thread","tid",
        "request_id","req_id","correlation_id","trace_id","x_request_id",
        "user_id","uid","user.id","principal","client_ip","ip","remote_addr",
        "method","http_method","path","url","uri","status","http_status","latency","latency_ms","duration","elapsed"
    ])
    extras = {k: v for (k, v) in kvs if k not in mapped_keys}
    row["extras_json"] = json.dumps(extras, separators=(",",":")) if extras else "{}"

    return row

def emit_tsv_row(row):
    fields = [sanitize(row.get(col, "")) for col in HEADER]
    sys.stdout.write("\t".join(fields) + "\n")

def main():
    # Write header
    sys.stdout.write("\t".join(HEADER) + "\n")
    for line in sys.stdin:
        stripped = line.lstrip()
        parsed = None
        if stripped.startswith("{"):
            parsed = parse_json_line(line)
        if parsed is None:
            parsed = parse_text_line(line)
        emit_tsv_row(parsed)

if __name__ == "__main__":
    main()

8. Operational guidance

• Run as: python log_to_tsv.py < input.log > output.tsv
• For large files, consider:
  • Using buffered I/O (default in Python is sufficient for most cases).
  • Precompiling all regexes (done).
  • Avoiding per-line JSON dumps for empty extras (use "{}" constant as shown).
  • Parallelization: sharding input files and merging outputs; ensure identical headers.
• Monitoring and testing:
  • Create unit tests with representative samples for each log format you encounter.
  • Validate TSV with a reader that checks column counts and escaping.
  • Track parse failure rates (e.g., empty ts_iso or level) and revisit patterns accordingly.

This logic establishes a deterministic, streaming-safe conversion of diverse logs to TSV suitable for ingestion into data warehouses or lakehouse tables. Adjust field mappings to your domain, and keep extras_json for schema evolution.

以下は、JSONをYAMLへ安全かつ忠実に変換するための実務的な変換ロジックです。目的は「JSONの型・値の意味を保ち、YAMLの解釈による型誤推測を回避する」ことです。YAML 1.2準拠を前提に記述します。

前提

• JSONはYAML 1.2の厳密な部分集合であり、原則として型対応は1対1で可能。
• ただしYAMLパーサ（特にYAML 1.1系）による曖昧なスカラー解釈（例: yes/no, on/off のブーリアン解釈）を避けるため、文字列の引用規則を設ける。
• JSONのオブジェクトのキーは文字列のみ。YAMLのマッピングキーも文字列で出力する。
• 順序は入力順を維持（一般的なJSONパーサでは挿入順を保持できる）。

型マッピング規則

• Object（JSON）→ Mapping（YAML）
  • { "k": v } → k: v
  • インデントは2スペース推奨。
  • ドキュメント開始記号（---）は任意。複数ドキュメント扱いが不要なら省略可。
• Array（JSON）→ Sequence（YAML）
  • [a, b] →
    • a
    • b
  • 順序はそのまま保持。
• String（JSON）→ Scalar（YAML）
  • 文字列は「プレーン」「ダブルクォート」「リテラルブロック（|）」のいずれかで表現。
  • 判定ロジックは後述。
• Number（JSON）→ Number（YAML）
  • 整数・浮動小数をそのまま文字列表現へ。
  • JSONにNaN/Infinityは存在しない前提（出力しない）。
• Boolean（JSON）→ Boolean（YAML）
  • true → true、false → false（小文字）
• Null（JSON）→ Null（YAML）
  • null → null（~は使わない方針）
• 日付/時刻
  • JSONでは文字列。YAML側で型推測されないよう文字列のまま出力（必要に応じて引用）。

文字列の引用判定ロジック（YAML 1.2を前提に安全運用）

• 以下のいずれかに該当する場合はダブルクォートで囲む（"..."）。
  • 先頭または末尾に空白がある
  • 改行を含まないが、次の文字を含む/先頭が危険文字
    • 「#」（コメント開始になりうる）
    • 「: 」（コロンに続くスペースはマッピング記号と衝突）
    • 先頭が「-」「?」「@」「`」「!」「*」「&」「%」「|」「>」
  • JSONの文字列値が次の曖昧トークンに完全一致（大小問わず）
    • "true", "false", "null", "~", "yes", "no", "on", "off"
    • 注: YAML 1.2では yes/no/on/off はブーリアンではないが、1.1系を避けるため安全に引用する
  • ISO8601風の日時文字列（例: 2024-01-01T12:00:00Z 等）はパーサ差異対策として引用推奨
  • 制御文字や非表示文字を含む（JSONのエスケープをYAMLのダブルクォート内で保持）
• 文字列に改行が含まれる場合はリテラルブロック（|）で出力し、改行を忠実に保持
  • 末尾改行の有無も正確に維持
• 上記に該当しない場合のみプレーンスカラー（引用なし）で出力
  • プレーン許容文字の目安: 英数字、ハイフン（-）、アンダースコア（_）、ドット（.）、スラッシュ（/）
  • 空文字はダブルクォートで "" とする

出力フォーマット規則

• インデントは2スペース
• シーケンスは「- 」で要素を列挙
• マッピングは「key: value」形式
• 必要に応じてドキュメント開始記号「---」を付与
• アンカー（&）やエイリアス（*）、タグ（!!）は生成しない（JSON互換の維持）

変換アルゴリズム（擬似コード）

• function to_yaml(value, indent=0):
  • switch type(value):
    • dict:
      • for (k, v) in insertion_order(value):
        • k_out = quote_key_if_needed(k) // キーも文字列。通常はプレーンだが危険時はダブルクォート
        • print(indent_spaces(indent) + k_out + ": " + inline_or_child(v, indent))
    • list:
      • for item in value:
        • print(indent_spaces(indent) + "- " + inline_or_child(item, indent))
    • string:
      • if contains_newline(value):
        • print(block_literal("|", value, indent)) // 改行をそのまま保持
      • else if needs_double_quote(value):
        • print(double_quoted(value)) // JSONエスケープはYAMLダブルクォートへ移植
      • else:
        • print(plain(value))
    • bool:
      • print("true" if value else "false")
    • null:
      • print("null")
    • number:
      • print(number_to_string(value)) // 整数/浮動小数を文字列化（指数表記等はそのまま）
• function needs_double_quote(s):
  • if s == "" return true
  • if leading_or_trailing_space(s) return true
  • if contains_any(s, ["#"]) return true
  • if contains_substring(s, ": ") return true
  • if starts_with_any(s, ["-", "?", "@", "`", "!", "*", "&", "%", "|", ">"]) return true
  • if is_ambiguous_token_ci(s, ["true","false","null","~","yes","no","on","off"]) return true
  • if looks_like_iso8601_timestamp(s) return true
  • if contains_control_char(s) return true
  • return false

Python実装例（ruamel.yamlを使用、YAML 1.2想定）

• ポイント
  • ruamel.yaml.scalarstring の LiteralScalarString（|）、DoubleQuotedScalarString（"）を使ってスタイルを明示。
  • 再帰的にJSON値を走査してスタイル化したオブジェクトへ変換後、YAMLへダンプ。
• コード
  • from json import loads
  • from ruamel.yaml import YAML
  • from ruamel.yaml.scalarstring import LiteralScalarString, DoubleQuotedScalarString
  • def convert_json_to_yaml_str(json_str):
    • data = loads(json_str)
    • def style_string(s):
      • if "\n" in s:
        • return LiteralScalarString(s) # 改行保持
      • if needs_double_quote(s):
        • return DoubleQuotedScalarString(s)
      • return s # プレーン
    • def transform(node):
      • if isinstance(node, dict):
        • return { (style_string(k) if isinstance(k, str) else k): transform(v) for k, v in node.items() }
      • if isinstance(node, list):
        • return [ transform(x) for x in node ]
      • if isinstance(node, str):
        • return style_string(node)
      • return node # bool, None, number はそのまま
    • yaml = YAML()
    • yaml.version = (1, 2)
    • yaml.indent(mapping=2, sequence=2, offset=0)
    • out = transform(data)
    • import io
    • buf = io.StringIO()
    • yaml.dump(out, buf)
    • return buf.getvalue()
  • 上記の needs_double_quote(s) は前述ロジックをそのままPythonで実装。

検証・テスト観点

• 型保持テスト
  • JSONのboolean/null/numberがYAMLで同型として再解釈されることを確認。
• 文字列の安全性テスト
  • "yes", "on", "#comment", ": value" 等が文字列として保持される（引用される）ことを確認。
  • 改行文字列がリテラルブロックで出力され、改行・末尾改行が一致することを確認。
• 構造・順序
  • 配列順序・オブジェクト挿入順を保持。
• エッジケース
  • 空文字列、先頭/末尾空白、長大文字列（折り畳み不要方針なら幅制限を外す）、Unicode絵文字・サロゲートペア。

このロジックにより、JSONの意味を損なわず、YAMLのパーサ差異による型誤解釈を最小化した安全な変換が可能です。