OpenDistro ELK集群无Logstash从IP获取Geo信息的方案问询

以下是三种经过验证的可行方案，可根据你的业务场景选择：

方案1：Elasticsearch内置Ingest Pipeline处理（最推荐） #

该方案完全依托ES原生能力，无需引入额外组件，也不需要修改应用写入逻辑，写入时即可实时完成Geo解析。
注意：使用前确认ES集群已安装ingest-geoip插件，高版本ES默认自带，缺失可通过ES内置插件管理工具安装

• 第一步：准备MMDB格式的GeoIP城市库，放到ES所有ingest节点的config/ingest-geoip目录下
• 第二步：创建Geo解析的Ingest Pipeline，示例如下：

PUT _ingest/pipeline/geoip_lookup
{
  "description": "原始IP转Geo地理信息",
  "processors": [
    {
      "geoip": {
        "field": "client_ip", // 替换为你数据中存储原始IP的字段名
        "target_field": "geo_info", // 解析后Geo信息的存储字段名
        "properties": ["country_name", "region_name", "city_name", "location", "ip"]
      }
    }
  ]
}

• 第三步：关联Pipeline到索引，推荐直接设置为索引默认Pipeline，无需修改应用写入请求：

PUT 你的索引名/_settings
{
  "index.default_pipeline": "geoip_lookup"
}

设置完成后所有新写入该索引的数据，都会自动完成IP到Geo信息的转换。

方案2：应用侧直接嵌入GeoIP解析 #

如果不希望调整ES侧配置，可以直接在写入ES的业务应用中集成GeoIP解析能力，写入前完成转换。

• 各主流开发语言都有成熟的MMDB格式解析库：Java可使用maxmind-db、Python可使用geoip2、Go可使用geoip2-golang
• 只需在应用本地部署GeoIP库文件，调用API解析IP得到Geo信息后，和业务数据一起写入ES即可，解析逻辑非常轻量，对应用性能影响可以忽略。

方案3：存量数据后置批量处理脚本 #

如果你的ES中已经存在未携带Geo信息的历史数据，可以使用下面的Python脚本批量补全：
首先安装依赖：
pip install elasticsearch geoip2
可运行脚本示例：

from elasticsearch import Elasticsearch, helpers
import geoip2.database

# 按需修改以下配置项
ES_ADDRESS = "http://你的ES服务地址:9200"
TARGET_INDEX = "你要处理的索引名"
RAW_IP_FIELD = "client_ip" # 存储原始IP的字段名
GEO_TARGET_FIELD = "geo_info" # 要写入Geo信息的目标字段
GEO_DB_PATH = "./GeoLite2-City.mmdb" # 本地GeoIP库文件路径

# 初始化客户端
es_client = Elasticsearch(ES_ADDRESS)
geo_reader = geoip2.database.Reader(GEO_DB_PATH)

def build_update_action(doc):
    try:
        raw_ip = doc["_source"][RAW_IP_FIELD]
        geo_result = geo_reader.city(raw_ip)
        # 组装Geo信息
        geo_data = {
            "country_name": geo_result.country.name,
            "region_name": geo_result.subdivisions.most_specific.name,
            "city_name": geo_result.city.name,
            "location": {
                "lat": geo_result.location.latitude,
                "lon": geo_result.location.longitude
            }
        }
        # 返回ES更新操作结构
        return {
            "_op_type": "update",
            "_index": doc["_index"],
            "_id": doc["_id"],
            "doc": {
                GEO_TARGET_FIELD: geo_data
            }
        }
    except Exception:
        # 无效IP、内网IP等解析失败的情况直接跳过
        return None

# 批量扫描未处理的文档并更新
action_buffer = []
# 只扫描还没有Geo字段的文档，避免重复处理
for doc in helpers.scan(es_client, index=TARGET_INDEX, query={
    "query": {
        "bool": {
            "must_not": {
                "exists": {"field": GEO_TARGET_FIELD}
            }
        }
    }
}):
    update_action = build_update_action(doc)
    if update_action:
        action_buffer.append(update_action)
    # 每1000条批量提交一次，可根据ES负载调整大小
    if len(action_buffer) >= 1000:
        helpers.bulk(es_client, action_buffer)
        action_buffer = []
# 提交剩余的更新操作
if action_buffer:
    helpers.bulk(es_client, action_buffer)

geo_reader.close()

脚本使用注意：可通过调整批量提交的条数控制处理速度，避免给ES集群造成过大压力

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