Python连接和操作Elasticsearch的流程步骤

docker pull docker.elastic.co/elasticsearch/elasticsearch:7.10.0
docker run --name elasticsearch -d -p 9200:9200 -p 9300:9300 elasticsearch:7.10.0

pip install elasticsearch

from elasticsearch import Elasticsearch

# 连接本地的 Elasticsearch 实例
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])

# 检查连接是否成功
if es.ping():
    print("连接成功！")
else:
    print("连接失败！")

es = Elasticsearch([{'host': '远程IP地址', 'port': 9200}])

# 检查连接
if es.ping():
    print("连接成功！")
else:
    print("连接失败！")

# 创建一个索引
index_name = "my_index"
response = es.indices.create(index=index_name, ignore=400)  # ignore 400 错误是因为索引已存在
print(response)

mapping = {
    "mappings": {
        "properties": {
            "name": {"type": "text"},
            "age": {"type": "integer"},
            "timestamp": {"type": "date"}
        }
    }
}

response = es.indices.create(index="my_index_with_mapping", body=mapping, ignore=400)
print(response)

document = {
    "name": "John Doe",
    "age": 29,
    "timestamp": "2024-12-24T10:00:00"
}

# 插入数据到索引
response = es.index(index="my_index", document=document)
print(response)

from elasticsearch.helpers import bulk

# 批量插入数据
actions = [
    {
        "_op_type": "index",  # 操作类型，可以是 index、update、delete
        "_index": "my_index",
        "_source": {
            "name": "Alice",
            "age": 30,
            "timestamp": "2024-12-24T12:00:00"
        }
    },
    {
        "_op_type": "index",
        "_index": "my_index",
        "_source": {
            "name": "Bob",
            "age": 35,
            "timestamp": "2024-12-24T12:05:00"
        }
    }
]

# 执行批量插入
success, failed = bulk(es, actions)
print(f"成功插入 {success} 条，失败 {failed} 条")

response = es.search(index="my_index", body={
    "query": {
        "match_all": {}  # 查询所有文档
    }
})
print(response)

response = es.search(index="my_index", body={
    "query": {
        "match": {
            "name": "John Doe"  # 查找name字段为"John Doe"的文档
        }
    }
})
print(response)

response = es.search(index="my_index", body={
    "query": {
        "bool": {
            "must": [
                {"match": {"name": "Alice"}},
                {"range": {"age": {"gte": 25}}}
            ],
            "filter": [
                {"term": {"timestamp": "2024-12-24T12:00:00"}}
            ]
        }
    }
})
print(response)

response = es.search(index="my_index", body={
    "query": {
        "range": {
            "age": {
                "gte": 30
            }
        }
    }
})
print(response)

document_id = "1"  # 假设这是我们要更新文档的 ID

update_doc = {
    "doc": {
        "age": 31
    }
}

response = es.update(index="my_index", id=document_id, body=update_doc)
print(response)

document_id = "1"  # 假设这是我们要删除文档的 ID
response = es.delete(index="my_index", id=document_id)
print(response)

response = es.search(index="my_index", body={
    "size": 0,  # 不返回文档，只返回聚合结果
    "aggs": {
        "average_age": {
            "avg": {
                "field": "age"
            }
        },
        "age_range": {
            "range": {
                "field": "age",
                "ranges": [
                    {"to": 30},
                    {"from": 30, "to": 40},
                    {"from": 40}
                ]
            }
        }
    }
})

# 打印聚合结果
print(response['aggregations'])

response = es.indices.delete(index="my_index", ignore=[400, 404])
print(response)

# 创建索引别名
response = es.indices.put_alias(index="my_index", name="my_index_alias")
print(response)

# 使用别名查询
response = es.search(index="my_index_alias", body={
    "query": {
        "match_all": {}
    }
})
print(response)

template = {
    "index_patterns": ["log-*"],  # 匹配所有以 log- 开头的索引
    "mappings": {
        "properties": {
            "timestamp": {"type": "date"},
            "log_level": {"type": "keyword"}
        }
    }
}

response = es.indices.put_template(name="log_template", body=template)
print(response)