数据仓库ETL日志监控：ELK Stack搭建+日志分析+告警配置实战

（注：实际发布时请替换为ELK Stack架构图）

前言：数据仓库的"黑匣子"困境与ELK的破局之道

在当今数据驱动的企业环境中，数据仓库作为核心决策支持系统，承载着从业务系统抽取、转换、加载数据的关键任务。ETL（Extract-Transform-Load）流程作为数据仓库的"血液"，其稳定性和可靠性直接关系到整个数据平台的可用性。

然而，在实际生产环境中，ETL作业往往面临以下挑战：

• 黑匣子困境：ETL作业执行过程不透明，出现问题难以定位
• 故障发现滞后：作业失败后往往需要人工检查才能发现，导致数据延迟
• 性能瓶颈：随着数据量增长，ETL作业性能问题日益突出
• 缺乏全局视角：多个ETL工具和作业分散管理，难以形成统一监控视图

根据Gartner 2023年数据管理调查报告，68%的数据仓库项目延期或超预算是由于ETL流程问题导致，而其中83%的问题可以通过完善的日志监控系统提前发现或快速解决。

ELK Stack（Elasticsearch, Logstash, Kibana）作为开源日志管理和分析的事实标准，为解决上述问题提供了强大而灵活的解决方案。本文将深入探讨如何构建一个专业的ETL日志监控系统，从环境搭建到高级分析，全方位覆盖ELK Stack在数据仓库监控中的应用。

第一章：ELK Stack核心概念与架构解析

1.1 ELK Stack组件解析

ELK Stack由三个核心组件构成，它们协同工作形成完整的日志处理流水线：

Elasticsearch：分布式搜索与分析引擎

Elasticsearch是一个基于Lucene的分布式开源搜索引擎，专为日志和时间序列数据的快速存储、搜索和分析而优化。其核心特性包括：

• 分布式架构：自动分片和副本机制，提供高可用性和水平扩展能力
• 近实时搜索：数据写入后毫秒级可搜索
• 结构化与非结构化数据支持：灵活处理各种格式的日志数据
• 强大的聚合分析：支持复杂的统计分析和数据挖掘操作

Elasticsearch的底层数据结构是倒排索引，这使其在全文搜索场景下性能卓越。对于ETL日志这种半结构化数据，Elasticsearch的动态映射功能可以自动识别字段类型，大大简化了数据接入流程。

Logstash：日志收集与处理管道

Logstash是一个开源的服务器端数据处理管道，能够从多个来源采集数据，转换数据，然后将数据发送到指定的存储库。其核心价值在于：

• 多源采集：支持从文件、数据库、消息队列等多种来源收集日志
• 丰富的过滤器：提供超过200种内置过滤器，满足各种日志处理需求
• 数据转换能力：支持字段提取、类型转换、格式转换等操作
• 可扩展性：通过插件机制可以轻松扩展其功能

对于ETL日志监控而言，Logstash的关键作用是将不同ETL工具（如Informatica, DataStage, Talend, Flink, Spark等）产生的异构日志标准化，为后续分析奠定基础。

Kibana：数据可视化与交互平台

Kibana是Elasticsearch的可视化和交互界面，提供了强大的数据探索、分析和可视化能力。其主要功能包括：

• 丰富的图表类型：支持折线图、柱状图、饼图、热力图等多种可视化方式
• 交互式仪表盘：可以创建自定义仪表盘，实时监控关键指标
• 高级搜索与过滤：基于Elasticsearch查询语法，快速定位问题
• 告警功能：支持基于日志内容和指标的告警配置

在ETL监控场景中，Kibana允许数据工程师和运维人员通过直观的界面监控ETL作业状态、性能趋势和异常情况。

1.2 ELK Stack工作原理与数据流

ELK Stack的工作流程可以概括为以下几个阶段，形成一个完整的日志处理流水线：

数据流详细解析：

1. 日志产生：ETL工具（如Informatica PowerCenter, Apache NiFi等）在执行过程中生成各类日志文件或输出日志事件

2. 日志收集：Logstash通过适当的输入插件（如file, beats, jdbc等）收集这些日志数据

3. 日志处理：Logstash应用一系列过滤器（如grok, mutate, date等）对原始日志进行解析、清洗和转换，提取关键字段（如作业名称、步骤ID、执行时间、状态、错误码等）

4. 日志存储：处理后的结构化日志数据被发送到Elasticsearch，存储在预先定义或动态创建的索引中

5. 日志索引：Elasticsearch对日志数据建立索引，优化查询性能和聚合分析能力

6. 日志可视化与分析：Kibana连接到Elasticsearch，提供交互式查询、报表和仪表盘，使用户能够实时监控ETL作业状态

7. 告警与通知：基于预定义的规则，当检测到异常情况（如作业失败、执行超时等）时，系统触发告警并通过邮件、短信或企业IM工具通知相关人员

1.3 ELK vs 其他日志解决方案对比分析

在选择日志监控解决方案时，ELK Stack并非唯一选择。下表对比了几种主流日志管理方案的优缺点：

解决方案	优势	劣势	适用场景
ELK Stack	开源免费、高度可定制、生态完善、社区活跃	部署维护复杂、资源消耗较高、需要专业知识	中大型企业、有技术能力的团队、定制化需求高
Splunk	开箱即用、易于上手、功能全面、技术支持完善	商业许可昂贵、扩展性受限	预算充足的企业、需要快速部署、对支持要求高
Graylog	专为日志设计、架构简洁、操作简单	插件生态不如ELK丰富、高级功能有限	中小型企业、专注日志功能、追求简单部署
Loki(Promtail+Grafana)	轻量级、资源消耗低、与Prometheus集成好	日志查询能力有限、相对新兴	Kubernetes环境、云原生应用、与监控系统集成
ELK+Beats	增强数据采集能力、降低Logstash负载	架构更复杂、需要管理更多组件	大规模日志采集、多源异构日志场景

对于数据仓库ETL日志监控场景，ELK Stack提供了最佳的灵活性和功能性平衡，尤其是其强大的全文搜索和聚合分析能力，使其成为复杂ETL环境下日志分析的理想选择。

第二章：环境搭建与基础配置

2.1 硬件与软件环境要求

在开始部署ELK Stack之前，需要确保环境满足以下最低要求：

硬件要求

组件	CPU	内存	存储	网络
Elasticsearch	4核+	16GB+	100GB+ SSD	千兆网卡
Logstash	2核+	8GB+	50GB+	千兆网卡
Kibana	2核+	4GB+	20GB+	千兆网卡

性能优化建议：

• Elasticsearch的性能对内存非常敏感，建议分配系统可用内存的50%给Elasticsearch，但不超过31GB（JVM堆内存超过31GB会导致指针压缩失效）
• 日志存储建议使用SSD，以获得更好的I/O性能
• 生产环境中，Elasticsearch应至少部署3个节点以确保高可用性

软件要求

组件	支持的操作系统	依赖软件
Elasticsearch	Linux, Windows, macOS	Java 8/11
Logstash	Linux, Windows, macOS	Java 8/11
Kibana	Linux, Windows, macOS	Node.js (内置)

本文将以CentOS 7.9操作系统为例，部署ELK Stack 8.10.4版本（最新稳定版）。

2.2 Elasticsearch安装与配置

2.2.1 安装Java环境

Elasticsearch需要Java运行环境，推荐使用Adoptium Temurin (原AdoptOpenJDK)：

# 添加Adoptium仓库
rpm --import https://packages.adoptium.net/artifactory/api/gpg/key/public

cat > /etc/yum.repos.d/adoptium.repo << EOF
[Adoptium]
name=Adoptium
baseurl=https://packages.adoptium.net/artifactory/rpms/\$releasever/\$basearch
enabled=1
gpgcheck=1
gpgkey=https://packages.adoptium.net/artifactory/api/gpg/key/public
EOF

# 安装Java 11
yum install -y temurin-11-jdk

# 验证安装
java -version

2.2.2 安装Elasticsearch

# 导入Elasticsearch GPG密钥
rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

# 添加Elasticsearch仓库
cat > /etc/yum.repos.d/elasticsearch.repo << EOF
[elasticsearch]
name=Elasticsearch repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF

# 安装Elasticsearch
yum install -y elasticsearch

# 启动并设置开机自启
systemctl daemon-reload
systemctl enable --now elasticsearch

2.2.3 核心配置详解

Elasticsearch的主配置文件位于/etc/elasticsearch/elasticsearch.yml。对于ETL日志监控场景，以下配置项尤为重要：

# 集群名称，同一集群中的节点必须使用相同名称
cluster.name: etl-monitor-cluster

# 节点名称，每个节点应具有唯一名称
node.name: es-node-1

# 数据和日志存储路径
path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch

# 网络配置，设置绑定地址和端口
network.host: 0.0.0.0  # 生产环境应指定具体IP
http.port: 9200

# 发现和集群形成配置
discovery.seed_hosts: ["192.168.1.101", "192.168.1.102", "192.168.1.103"]  # 所有节点的IP
cluster.initial_master_nodes: ["es-node-1", "es-node-2", "es-node-3"]  # 初始主节点

# 安全配置 (Elasticsearch 8.x默认启用)
xpack.security.enabled: true
xpack.security.enrollment.enabled: true
xpack.security.http.ssl:
  enabled: true
  keystore.path: certs/http.p12
xpack.security.transport.ssl:
  enabled: true
  verification_mode: certificate
  keystore.path: certs/transport.p12
  truststore.path: certs/transport.p12

# 性能优化配置
indices.fielddata.cache.size: 20%  # 字段数据缓存大小
indices.queries.cache.size: 10%    # 查询缓存大小
thread_pool.write.queue_size: 1000 # 写入线程池队列大小

首次启动与安全配置：
Elasticsearch 8.x默认启用了安全功能，首次启动后需要设置内置用户密码：

/usr/share/elasticsearch/bin/elasticsearch-setup-passwords interactive

按照提示为elastic、kibana、logstash_system等用户设置强密码

2.2.4 验证安装

# 验证Elasticsearch服务状态
systemctl status elasticsearch

# 通过API验证节点状态
curl -X GET "https://localhost:9200/_cluster/health?pretty" -u elastic:your_password -k

成功安装后，应看到类似以下的响应：

{
  "cluster_name" : "etl-monitor-cluster",
  "status" : "green",  // green表示集群健康
  "timed_out" : false,
  "number_of_nodes" : 1,
  "number_of_data_nodes" : 1,
  "active_primary_shards" : 0,
  "active_shards" : 0,
  "relocating_shards" : 0,
  "initializing_shards" : 0,
  "unassigned_shards" : 0,
  "delayed_unassigned_shards" : 0,
  "number_of_pending_tasks" : 0,
  "number_of_in_flight_fetch" : 0,
  "task_max_waiting_in_queue_millis" : 0,
  "active_shards_percent_as_number" : 100.0
}

2.3 Logstash安装与配置

2.3.1 安装Logstash

# 安装Logstash (如果已添加Elastic仓库)
yum install -y logstash

# 启动并设置开机自启
systemctl daemon-reload
systemctl enable --now logstash

2.3.2 Logstash核心配置结构

Logstash配置文件采用分段结构，主要包含三个部分：

input {
  # 输入插件配置 - 从哪里收集数据
}

filter {
  # 过滤器配置 - 如何处理和转换数据
}

output {
  # 输出插件配置 - 将处理后的数据发送到哪里
}

2.3.3 针对ETL日志的基础配置示例

以下是一个处理ETL日志的基础Logstash配置示例（保存为/etc/logstash/conf.d/etl-log-pipeline.conf）：

input {
  # 从文件收集ETL日志
  file {
    path => "/var/log/etl/*.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"  # 开发环境使用，生产环境应保留sincedb
    tags => ["etl", "file"]
  }
  
  # 从TCP端口接收日志
  tcp {
    port => 5000
    tags => ["etl", "tcp"]
  }
}

filter {
  # 识别不同ETL工具的日志格式
  if "informatica" in [tags] {
    # Informatica ETL日志解析
    grok {
      match => { "message" => "%{TIMESTAMP_ISO8601:log_time} \[%{DATA:thread}\] %{LOGLEVEL:loglevel} %{DATA:component} - %{GREEDYDATA:message}" }
      add_field => { "etl_tool" => "informatica" }
    }
  } else if "datastage" in [tags] {
    # DataStage ETL日志解析
    grok {
      match => { "message" => "%{TIMESTAMP_ISO8601:log_time} %{DATA:project}\.%{DATA:job_name}\.%{DATA:stage_name} %{LOGLEVEL:loglevel}: %{GREEDYDATA:message}" }
      add_field => { "etl_tool" => "datastage" }
    }
  } else {
    # 默认ETL日志解析
    grok {
      match => { "message" => "%{TIMESTAMP_ISO8601:log_time} %{LOGLEVEL:loglevel} \[%{DATA:job_id}\] %{DATA:job_name}: %{GREEDYDATA:message}" }
      add_field => { "etl_tool" => "unknown" }
    }
  }
  
  # 日期处理
  date {
    match => [ "log_time", "yyyy-MM-dd HH:mm:ss", "ISO8601" ]
    target => "@timestamp"  # 覆盖默认时间戳
  }
  
  # 提取ETL作业状态
  if "successfully completed" in [message] {
    mutate {
      add_field => { "job_status" => "success" }
    }
  } else if "failed with error" in [message] {
    mutate {
      add_field => { "job_status" => "failed" }
      add_tag => [ "error", "job_failure" ]
      # 提取错误代码
      gsub => [ "message", ".*error code (\d+).*", "error_code=\1" ]
    }
    # 提取错误代码
    grok {
      match => { "message" => "error code (%{NUMBER:error_code:int})" }
      tag_on_failure => [ "no_error_code" ]
    }
  }
  
  # 移除不需要的字段
  mutate {
    remove_field => [ "log_time", "path", "host" ]
  }
}

output {
  # 输出到Elasticsearch
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "etl-logs-%{+YYYY.MM.dd}"  # 按日期创建索引
    user => "logstash_system"
    password => "your_logstash_password"
    ssl => true
    ssl_certificate_verification => false  # 开发环境，生产环境应启用证书验证
    # 文档ID配置
    document_id => "%{[@timestamp]}-%{job_id}"
  }
  
  # 同时输出到控制台（开发调试用）
  stdout {
    codec => rubydebug
  }
}

2.3.4 配置验证与服务管理

# 验证Logstash配置
/usr/share/logstash/bin/logstash --config.test_and_exit -f /etc/logstash/conf.d/etl-log-pipeline.conf

# 重新加载配置（Logstash 7.6+支持）
curl -X POST "http://localhost:9600/_node/hot_threads/reload"

# 或重启Logstash服务
systemctl restart logstash

# 查看Logstash日志
tail -f /var/log/logstash/logstash-plain.log

2.4 Kibana安装与基础配置

2.4.1 安装Kibana

# 安装Kibana (如果已添加Elastic仓库)
yum install -y kibana

# 启动并设置开机自启
systemctl daemon-reload
systemctl enable --now kibana

2.4.2 Kibana核心配置

Kibana的主配置文件位于/etc/kibana/kibana.yml：

# 服务器配置
server.port: 5601
server.host: "0.0.0.0"  # 生产环境应指定具体IP

# Elasticsearch连接配置
elasticsearch.hosts: ["https://localhost:9200"]
elasticsearch.username: "kibana"
elasticsearch.password: "your_kibana_password"

# Elasticsearch SSL配置
elasticsearch.ssl.verificationMode: none  # 开发环境，生产环境应设为full

# Kibana索引配置
kibana.index: ".kibana"

# 日志配置
logging.dest: /var/log/kibana/kibana.log
logging.verbose: false

# 国际化配置
i18n.locale: "zh-CN"  # 设置为中文界面

2.4.3 验证安装与访问Kibana

# 验证Kibana服务状态
systemctl status kibana

# 查看Kibana日志
tail -f /var/log/kibana/kibana.log

打开浏览器访问http://your_server_ip:5601，使用elastic用户和之前设置的密码登录Kibana。

首次登录后，Kibana会引导完成初始设置，包括添加索引模式等。

2.5 Filebeat安装与配置（轻量级日志采集）

对于大规模日志采集场景，直接使用Logstash可能会成为性能瓶颈。Filebeat作为轻量级日志采集器，可以替代Logstash的输入功能，提高采集效率并降低资源消耗。

2.5.1 安装Filebeat

# 添加Elastic仓库（如果尚未添加）
rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
cat > /etc/yum.repos.d/elastic.repo << EOF
[elastic-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF

# 安装Filebeat
yum install -y filebeat

# 启动并设置开机自启
systemctl daemon-reload
systemctl enable --now filebeat

2.5.2 Filebeat配置详解

Filebeat配置文件位于/etc/filebeat/filebeat.yml，以下是采集ETL日志的专用配置：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/etl/informatica/*.log
  tags: ["etl", "informatica"]
  fields:
    etl_tool: informatica
  # 多行日志处理
  multiline.type: pattern
  multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'  # 以日期开头的行为新日志行
  multiline.negate: true
  multiline.match: after

- type: log
  enabled: true
  paths:
    - /var/log/etl/datastage/*.log
  tags: ["etl", "datastage"]
  fields:
    etl_tool: datastage
  # 自定义日志解析
  processors:
    - add_patterns_path:
        path: /etc/filebeat/patterns.d
    - dissect:
        tokenizer: "%{log_time} %{project}.%{job_name}.%{stage_name} %{loglevel}: %{message}"
        field: "message"
        target_prefix: "ds"

# 输出配置 - 直接发送到Elasticsearch
# output.elasticsearch:
#   hosts: ["https://localhost:9200"]
#   username: "elastic"
#   password: "your_elastic_password"
#   ssl:
#     verification_mode: none
#   index: "etl-logs-%{+YYYY.MM.dd}"

# 输出配置 - 发送到Logstash进行进一步处理
output.logstash:
  hosts: ["localhost:5044"]
  ssl:
    verification_mode: none

# 索引生命周期管理
setup.ilm.enabled: true
setup.ilm.rollover_alias: "etl-logs"
setup.ilm.pattern: "{now/d}-000001"
setup.ilm.policy_name: "etl-logs-policy"

# Kibana仪表盘设置
setup.kibana:
  host: "http://localhost:5601"
  username: "elastic"
  password: "your_elastic_password"

# 启用内置仪表盘
setup.dashboards.enabled: true
setup.dashboards.index: "etl-logs-*"

2.5.3 自定义Grok模式

创建自定义Grok模式文件/etc/filebeat/patterns.d/etl：

# Informatica ETL日志模式
INFORMATICA_LOG %{TIMESTAMP_ISO8601:log_time} \[%{DATA:thread}\] %{LOGLEVEL:loglevel} %{DATA:component} - %{GREEDYDATA:message}

# DataStage ETL日志模式
DATASTAGE_LOG %{TIMESTAMP_ISO8601:log_time} %{DATA:project}\.%{DATA:job_name}\.%{DATA:stage_name} %{LOGLEVEL:loglevel}: %{GREEDYDATA:message}

# ETL错误模式
ETL_ERROR error code %{NUMBER:error_code:int}: %{GREEDYDATA:error_message}

2.5.4 配置Filebeat模块（可选）

Filebeat提供了多种预定义模块，可以简化常见日志源的配置：

# 查看可用模块
filebeat modules list

# 启用系统模块（示例）
filebeat modules enable system

# 配置模块
vi /etc/filebeat/modules.d/system.yml

# 加载模块配置并启动
filebeat setup
systemctl restart filebeat

2.6 分布式部署架构与最佳实践

对于生产环境，特别是大规模ETL环境，单节点ELK部署无法满足性能和可用性要求。以下是几种常见的分布式部署架构：

2.6.1 小规模分布式架构（3节点）

这种架构适用于中小规模环境，特点是：

• 3个Elasticsearch节点，兼具主节点和数据节点角色
• 单Logstash实例或多个独立Logstash实例
• Filebeat部署在各ETL服务器上采集日志

2.6.2 大规模集群架构

大规模集群架构特点：

• 专用的协调节点、主节点、数据节点和可能的协调节点
• 多个Logstash节点，前端配置负载均衡
• 完整的监控和告警系统
• 数据备份与恢复机制

2.6.3 容器化部署（Docker/Kubernetes）

随着容器技术的普及，ELK Stack也可以部署在容器环境中：

Docker Compose示例:

version: '3.8'

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.10.4
    container_name: elasticsearch
    environment:
      - cluster.name=etl-monitor-cluster
      - node.name=es-node-1
      - discovery.type=single-node
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
      - xpack.security.enabled=false
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - esdata:/usr/share/elasticsearch/data
    ports:
      - "9200:9200"
    networks:
      - elk-network

  logstash:
    image: docker.elastic.co/logstash/logstash:8.10.4
    container_name: logstash
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline
      - ./logstash/config:/usr/share/logstash/config
    ports:
      - "5000:5000"
      - "9600:9600"
    environment:
      - "LS_JAVA_OPTS=-Xms256m -Xmx256m"
    depends_on:
      - elasticsearch
    networks:
      - elk-network

  kibana:
    image: docker.elastic.co/kibana/kibana:8.10.4
    container_name: kibana
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    ports:
      - "5601:5601"
    depends_on:
      - elasticsearch
    networks:
      - elk-network

networks:
  elk-network:
    driver: bridge

volumes:
  esdata:
    driver: local

2.6.4 分布式部署最佳实践

1. Elasticsearch集群最佳实践：

   • 生产环境至少3个节点
   • 合理规划分片策略，避免过度分片
   • 启用索引生命周期管理(ILM)
   • 定期备份数据
   • 监控集群健康和性能指标

2. Logstash性能优化：

   • 适当增加worker数量和批处理大小
   • 使用持久化队列避免数据丢失
   • 避免在Logstash中进行复杂计算，尽量在Elasticsearch中完成
   • 考虑使用Filebeat直接发送到Elasticsearch减轻Logstash负担

3. 网络与安全最佳实践：

   • 所有组件间通信启用SSL/TLS加密
   • 使用专用用户账户和最小权限原则
   • 配置防火墙限制访问
   • 敏感日志数据考虑加密存储

4. 监控与维护：

   • 部署Elasticsearch监控（使用Metricbeat和内置监控）
   • 设置关键指标告警
   • 制定明确的索引清理和归档策略
   • 定期更新ELK Stack版本

第三章：ETL日志采集与处理管道设计

3.1 ETL日志类型与格式解析

ETL过程会产生多种类型的日志，了解这些日志的特点是构建有效监控系统的基础。

3.1.1 ETL日志分类

根据日志产生的阶段和内容，ETL日志可分为以下几类：

1. 作业执行日志：记录整个ETL作业的执行过程，包括启动、各阶段执行、完成或失败等关键事件
2. 数据转换日志：详细记录数据转换过程中的操作，如数据清洗、字段映射、聚合计算等
3. 数据加载日志：记录数据写入目标数据仓库的过程，包括记录数、成功数、失败数等统计信息
4. 错误日志：专门记录ETL过程中发生的错误和异常信息
5. 性能日志：记录ETL作业各环节的执行时间、资源消耗等性能指标
6. 审计日志：记录ETL作业的访问、修改等操作，满足合规性要求

3.1.2 主流ETL工具日志格式

不同ETL工具产生的日志格式差异较大，以下是几种常见ETL工具的日志示例：

Informatica PowerCenter日志：

2023-10-25 08:30:15 [WRT_8229] INFO  Writer_1_1_1 - Starting to write to target table [CUSTOMER].
2023-10-25 08:30:17 [WRT_8164] INFO  Writer_1_1_1 - Target load complete. Total rows written to target [CUSTOMER]: [12500].
2023-10-25 08:30:18 [SESSION.1000] INFO  Session - Session [s_Customer_Load] completed successfully.

IBM DataStage日志：

2023-10-25 08:45:02 DSJobController.JobControl (DSRunJob) Job job_Customer_ETL has finished, status = 0 (Finished OK)
2023-10-25 08:45:02 DSJobController.JobControl (Transformer_1) Stage "Transformer_1" processed 12500 input rows.
2023-10-25 08:45:03 DSJobController.JobControl (Sequential_File_1) Read 12500 records from source file.

Talend日志：

2023-10-25 09:00:01 [main] INFO  customer_etl - Starting job customer_etl at 2023-10-25 09:00:01
2023-10-25 09:00:05 [main] INFO  tOracleInput_1 - Extracting 12500 rows from table CUSTOMER
2023-10-25 09:00:10 [main] ERROR tMap_1 - Error converting data: Value 'N/A' cannot be converted to integer
2023-10-25 09:00:15 [main] INFO  customer_etl - Job customer_etl finished with status ERROR in 14s

Apache NiFi日志：

2023-10-25T09:15:00.123+0800 INFO [Timer-Driven Process Thread-1] o.a.n.p.standard.LogAttribute LogAttribute[id=123456] FlowFile UUID: 123e4567-e89b-12d3-a456-426614174000, Filename: customer_data.csv, Records Count: 12500
2023-10-25T09:15:02.456+0800 WARN [Timer-Driven Process Thread-1] o.a.n.p.standard.ValidateRecord ValidateRecord[id=7890ab] Record 153 contains invalid email address: 'user@domain'
2023-10-25T09:15:05.789+0800 INFO [Timer-Driven Process Thread-1] o.a.n.p.standard.PutDatabaseRecord PutDatabaseRecord[id=abcdef] Successfully inserted 12450 records into table CUSTOMER

3.1.3 ETL日志关键字段提取

无论何种ETL工具，以下关键字段对监控和分析至关重要：

字段类别	关键字段	描述
基本信息	timestamp, etl_tool, job_name, job_id, execution_id	日志时间戳、ETL工具类型、作业名称、作业ID、执行实例ID
执行状态	status, step_name, progress, start_time, end_time	作业状态、步骤名称、进度百分比、开始时间、结束时间
数据统计	source_records, processed_records, loaded_records, rejected_records	源记录数、处理记录数、加载记录数、拒绝记录数
错误信息	error_code, error_message, error_stage, stack_trace	错误代码、错误消息、错误发生阶段、堆栈跟踪
性能指标	duration, throughput, cpu_usage, memory_usage	持续时间、吞吐量、CPU使用率、内存使用率
环境信息	host, user, project, workflow_name	主机名、执行用户、项目名称、工作流名称

3.2 Logstash管道配置详解

Logstash提供了强大的日志处理能力，通过精心配置的管道可以将原始ETL日志转换为结构化数据。

3.2.1 多源日志采集配置

Logstash支持从多种来源采集日志，以下是ETL监控场景中常用的输入插件配置：

文件输入插件增强配置：

input {
  file {
    path => "/opt/informatica/powercenter/server/infa_shared/SessLogs/*.log"
    start_position => "end"
    sincedb_path => "/var/lib/logstash/sincedb/informatica"
    codec => multiline {
      pattern => "^%{TIMESTAMP_ISO8601}"
      negate => true
      what => "previous"
      max_lines => 1000
      timeout => 30s
    }
    tags => ["informatica", "sesslog", "etl"]
    type => "informatica-session"
    stat_interval => 10
    discover_interval => 30
    close_older => 3600
    ignore_older => 86400
  }
  
  file {
    path => "/opt/ibm/datastage/Projects/*/Logs/*.log"
    start_position => "end"
    sincedb_path => "/var/lib/logstash/sincedb/datastage"
    tags => ["datastage", "etl"]
    type => "datastage-job"
  }
}

数据库日志输入（适用于存储在数据库中的ETL日志）：

input {
  jdbc {
    jdbc_driver_library => "/usr/share/logstash/lib/ojdbc8.jar"
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
    jdbc_connection_string => "jdbc:oracle:thin:@//db-host:1521/etlrepo"
    jdbc_user => "etl_monitor"
    jdbc_password => "secure_password"
    schedule => "* * * * *"  # 每分钟执行一次查询
    statement => "SELECT * FROM etl_job_logs WHERE log_time > :sql_last_value"
    use_column_value => true
    tracking_column => "log_time"
    tracking_column_type => "timestamp"
    last_run_metadata_path => "/var/lib/logstash/lastrun/etl_job_logs"
    tags => ["db-log", "etl"]
    type => "database-etl-log"
    jdbc_fetch_size => 1000
  }
}

TCP输入（适用于ETL工具主动发送日志）：

input {
  tcp {
    port => 5000
    mode => "server"
    ssl_enable => true
    ssl_cert => "/etc/logstash/certs/server.crt"
    ssl_key => "/etc/logstash/certs/server.key"
    ssl_verify => false
    tags => ["tcp", "etl"]
    type => "tcp-etl-log"
    codec => json_lines
  }
}

3.2.2 高级过滤插件应用

Logstash的强大之处在于其丰富的过滤插件，以下是ETL日志处理中最常用的过滤插件配置：

Grok插件深度应用：

Grok是Logstash最强大的解析工具，使用正则表达式将非结构化日志转换为结构化数据。

filter {
  if "informatica" in [tags] {
    grok {
      match => { "message" => [
        # Informatica会话日志模式
        "%{TIMESTAMP_ISO8601:log_time} \[%{DATA:thread}\] %{LOGLEVEL:loglevel} %{DATA:component} - %{GREEDYDATA:message}",
        # Informatica错误日志模式
        "%{TIMESTAMP_ISO8601:log_time} \[%{DATA:thread}\] %{LOGLEVEL:loglevel} %{DATA:component} - ERROR \(%{NUMBER:error_code:int}\): %{GREEDYDATA:error_message}",
        # Informatica会话完成模式
        "%{TIMESTAMP_ISO8601:log_time} \[%{DATA:thread}\] %{LOGLEVEL:loglevel} %{DATA:component} - Session %{DATA:session_name} completed with status %{DATA:status}."
      ]}
      pattern_definitions => {
        "INFA_THREAD" => "([A-Za-z0-9_]+)"
        "INFA_COMPONENT" => "([A-Za-z0-9_]+)"
      }
      overwrite => ["message"]
      add_field => { "etl_tool" => "informatica" }
      tag_on_failure => ["_informatica_grok_failure"]
    }
  }
}

# 提取通用ETL指标
grok {
  match => { "message" => [
    "Total rows written to target \[%{DATA:target_table}\]: \[%{NUMBER:loaded_records:int}\]",
    "Source rows read: %{NUMBER:source_records:int}",
    "Rows rejected: %{NUMBER:rejected_records:int}",
    "Session \[%{DATA:session_name}\] completed with status %{DATA:status}"
  ]}
  tag_on_failure => ["_metric_grok_failure"]
}

日期处理与时间戳规范化：

filter {
  date {
    match => [ 
      "log_time", "yyyy-MM-dd HH:mm:ss", 
      "execution_start_time", "ISO8601",
      "start_time", "yyyy-MM-dd'T'HH:mm:ss.SSSZ"
    ]
    target => "@timestamp"
    timezone => "Asia/Shanghai"
    add_field => { "log_date" => "%{+YYYY-MM-dd}" }
    tag_on_failure => ["_dateparsefailure"]
  }
  
  # 计算执行持续时间
  if [start_time] and [end_time] {
    ruby {
      code => "
        require 'time'
        start = Time.parse(event.get('start_time'))
        end_t = Time.parse(event.get('end_time'))
        event.set('duration_seconds', end_t - start)
      "
      tag_on_failure => ["_duration_calc_failure"]
    }
  }
}

条件逻辑与字段操作：

filter {
  # 根据日志级别设置严重性
  mutate {
    add_field => { "severity" => 0 }
  }
  
  if [loglevel] == "ERROR" {
    mutate {
      update_field => { "severity" => 3 }
      add_tag => ["critical", "alert"]
    }
  } else if [loglevel] == "WARN" {
    mutate {
      update_field => { "severity" => 2 }
      add_tag => ["warning"]
    }
  } else if [loglevel] == "INFO" {
    mutate {
      update_field => { "severity" => 1 }
    }
  }
  
  # 重命名字段
  mutate {
    rename => { 
      "session_name" => "job_name"
      "workflow_id" => "execution_id"
    }
  }
  
  # 类型转换
  mutate {
    convert => { 
      "source_records" => "integer"
      "loaded_records" => "integer"
      "rejected_records" => "integer