如何用Logstash读取多CSV文件合并生成Elasticsearch统一数据视图

完全支持将两个独立的CSV数据集合并为包含CustomerID、CustomerName、ContactName、Country、CustomerCreateDate、OrderID、OrderDate全字段的统一数据视图，你可以根据当前数据入库进度选择以下两种落地方式：

方案1：Logstash入库阶段关联（推荐，查询性能最优） #

核心逻辑是先把客户维度数据加载到内存缓存，读取订单流时通过CustomerID匹配对应客户信息，拼接为完整记录后写入同一个ES索引，后续直接基于该索引创建数据视图即可。

• 配置加载逻辑时优先读取Customers.csv，全量解析后写入内存字典
• 读取Orders.csv逐行解析时，根据每行的CustomerID从字典中取出对应的客户属性，补全到当前记录
• 过滤掉初始化用的客户临时数据，仅把拼接完成的订单+客户全字段记录写入统一索引
  完整配置参考：

input {
  # 读取客户数据初始化缓存
  file {
    path => "/path/to/Customers.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null"
    mode => "read"
    exit_after_read => true
    tags => ["init_customer"]
  }
  # 读取订单数据做关联
  file {
    path => "/path/to/Orders.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null"
    tags => ["process_order"]
  }
}

filter {
  # 处理客户初始化数据，写入内存字典
  if "init_customer" in [tags] {
    csv {
      separator => ","
      columns => ["CustomerID","CustomerName","ContactName","Country","CustomerCreateDate"]
      skip_header => true
    }
    mutate {
      convert => { "CustomerID" => "integer" }
    }
    translate {
      field => "CustomerID"
      dictionary_path => "/tmp/customer_dict.json"
      refresh_interval => 300
      destination => "[@metadata][target]"
    }
    # 临时客户数据不写入ES，仅做缓存
    drop {}
  }

  # 处理订单数据，关联客户字段
  if "process_order" in [tags] {
    csv {
      separator => ","
      columns => ["OrderID","CustomerID","OrderDate"]
      skip_header => true
    }
    mutate {
      convert => {
        "OrderID" => "integer"
        "CustomerID" => "integer"
      }
    }
    # 从内存字典匹配客户信息
    translate {
      field => "CustomerID"
      dictionary_path => "/tmp/customer_dict.json"
      destination => "customer_match"
    }
    # 提取客户字段到顶层
    if [customer_match] {
      mutate {
        add_field => {
          "CustomerName" => "%{[customer_match][CustomerName]}"
          "ContactName" => "%{[customer_match][ContactName]}"
          "Country" => "%{[customer_match][Country]}"
          "CustomerCreateDate" => "%{[customer_match][CustomerCreateDate]}"
        }
        remove_field => ["customer_match", "@version", "tags", "host", "path", "@timestamp"]
      }
    }
    # 示例中CustomerID=37、77的订单无匹配客户，对应客户字段为空属于正常情况，不会抛错
  }
}

output {
  if "process_order" in [tags] {
    elasticsearch {
      hosts => "elasticsearch:9200"
      user => "logstash_internal"
      password => "${LOGSTASH_INTERNAL_PASSWORD}"
      index => "customer_orders_unified"
    }
  }
}

注意：如果客户数据后续会更新，保持refresh_interval => 300的配置即可，Logstash会每5分钟自动重载客户数据文件，无需重启服务。

方案2：ES层Enrich关联（无需重新导入原始CSV） #

如果你已经按照原有配置把客户、订单数据分别写入了customers和orders两个索引，不需要重新处理原始文件，直接用ES自带的Enrich能力即可完成关联：

1. 为客户索引创建Enrich匹配策略，指定CustomerID为关联键
2. 执行策略生成关联缓存
3. 创建Ingest Pipeline，配置关联处理器自动补全客户字段
4. 通过Reindex接口把订单数据经Pipeline处理后写入统一索引
   核心操作命令参考：

# 创建Enrich策略
PUT /_enrich/policy/customer_lookup
{
  "match": {
    "indices": "customers",
    "match_field": "CustomerID",
    "enrich_fields": ["CustomerName", "ContactName", "Country", "CustomerCreateDate"]
  }
}

# 执行策略生成缓存
POST /_enrich/policy/customer_lookup/_execute

# 创建关联用的Ingest Pipeline
PUT /_ingest/pipeline/join_customer_order
{
  "processors": [
    {
      "enrich": {
        "policy_name": "customer_lookup",
        "field": "CustomerID",
        "target_field": "customer_tmp",
        "max_matches": 1
      }
    },
    {
      "script": """
        if (ctx.customer_tmp != null) {
          ctx.CustomerName = ctx.customer_tmp.CustomerName;
          ctx.ContactName = ctx.customer_tmp.ContactName;
          ctx.Country = ctx.customer_tmp.Country;
          ctx.CustomerCreateDate = ctx.customer_tmp.CustomerCreateDate;
          ctx.remove('customer_tmp');
        }
      """
    }
  ]
}

# 重索引生成统一关联索引
POST /_reindex
{
  "source": {"index": "orders"},
  "dest": {
    "index": "customer_orders_unified",
    "pipeline": "join_customer_order"
  }
}

两种方案最终都会生成包含所有目标字段的customer_orders_unified索引，直接在Kibana中针对该索引创建数据视图即可满足需求。

内容的提问来源于stack exchange，提问作者Brian