网站首页PHP

elasticsearch库使用示例

使用Composer安装elasticsearch库
[root@localhost src]# composer require elasticsearch/elasticsearch
 
<?php
declare(strict_types=1);

require_once __DIR__ . '/elasticsearch/vendor/autoload.php';

use Elastic\Elasticsearch\ClientBuilder;
use Elastic\Elasticsearch\Exception\AuthenticationException;
use Elastic\Elasticsearch\Exception\ClientResponseException;
use Elastic\Elasticsearch\Exception\MissingParameterException;
use Elastic\Elasticsearch\Exception\ServerResponseException;

ini_set('display_errors', 'On');
error_reporting(-1);

function json_de(string $json): array
{
    try {
        $array = json_decode($json, true, 512, JSON_THROW_ON_ERROR);
    } catch (JsonException $e) {
        exit('JsonException: ' . $e->getMessage());
    }

    !is_array($array) && exit('json_decode() error.');

    return $array;
}

$hosts = ['localhost:9200']; // Elasticsearch主机地址
$index = 'my-index-0'; // Elasticsearch的索引相当于MySQL的数据库，即一个索引对应一个数据库


//========== 创建Elasticsearch客户端 ==========//
try {
    $client = ClientBuilder::create()->setHosts($hosts)->build();
} catch (AuthenticationException $e) {
    exit('创建Elasticsearch客户端失败：' . $e->getMessage());
}


//========== 获取Elasticsearch服务端信息 ==========//
try {
    $response = $client->info();
} catch (ClientResponseException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $array = $response->asArray();
    $versionNumber = (string)($array['version']['number'] ?? 'N/A');
    echo "Elasticsearch版本：$versionNumber" . PHP_EOL;
} else {
    echo "获取Elasticsearch服务端信息失败：$statusCode $reasonPhrase" . PHP_EOL;
}


echo PHP_EOL;
//========== 创建索引 ==========//
$params = ['index' => $index];
try {
    $response = $client->indices()->create($params);
} catch (ClientResponseException|MissingParameterException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $array = $response->asArray();
    $acknowledged = isset($array['acknowledged']) && $array['acknowledged'];
    $shardsAcknowledged = isset($array['shards_acknowledged']) && $array['shards_acknowledged'];
    $idx = isset($array['index']) ? (string)$array['index'] : '';
    $ok = $acknowledged && $shardsAcknowledged && $idx === $index;
    echo "创建索引[$index]" . ($ok ? '成功' : '失败') . PHP_EOL;
} elseif ($statusCode === 400) {
    echo "创建索引[$index]失败：该索引已存在" . PHP_EOL; // 状态码为400表示索引已存在
} else {
    echo "创建索引[$index]失败：$statusCode $reasonPhrase" . PHP_EOL;
}


echo PHP_EOL;
//========== 检查索引是否存在 ==========//
try {
    $response = $client->indices()->exists(['index' => $index]);
} catch (ClientResponseException|MissingParameterException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$exists = $response->getStatusCode() === 200 && $response->getReasonPhrase() === 'OK'; // 200=存在(OK)|404=不存在(Not Found)
echo "索引[$index]" . ($exists ? '存在' : '不存在') . PHP_EOL;


echo PHP_EOL;
//========== 获取所有索引名 ==========//
$params = ['index' => '_all', 'format' => 'json'];
try {
    $response = $client->cat()->indices($params);
} catch (ClientResponseException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$array = $response->asArray();
foreach ($array as $value) {
    $docsCount = (int)$value['docs.count']; // 该字段可能为null
    echo "索引名：{$value['index']} 记录数：$docsCount" . PHP_EOL;
}


echo PHP_EOL;
//========== 插入(或更新)记录 ==========//
$datas = [
    ['uid' => 1001, 'name' => '刘一', 'gender' => '男', 'birth' => 2001, 'text' => '北国风光，千里冰封，万里雪飘。望长城内外，惟余莽莽；大河上下，顿失滔滔。'],
    ['uid' => 1002, 'name' => '陈二', 'gender' => '女', 'birth' => 2002, 'text' => '独立寒秋，湘江北去，橘子洲头。看万山红遍，层林尽染；漫江碧透，百舸争流。'],
    ['uid' => 1003, 'name' => '张三', 'gender' => '男', 'birth' => 2003, 'text' => '孤馆灯青，野店鸡号，旅枕梦残。渐月华收练，晨霜耿耿；云山摛锦，朝露漙漙。'],
    ['uid' => 1004, 'name' => '李四', 'gender' => '女', 'birth' => 2004, 'text' => '叠嶂西驰，万马回旋，众山欲东。正惊湍直下，跳珠倒溅，小桥横截，缺月初弓。'],
    ['uid' => 1005, 'name' => '王五', 'gender' => '男', 'birth' => 2005, 'text' => '孤鹤归飞，再过辽天，换尽旧人。念累累枯冢，茫茫梦境，王侯蝼蚁，毕竟成尘。'],
    ['uid' => 1006, 'name' => '赵六', 'gender' => '女', 'birth' => 2006, 'text' => '斗酒彘肩，风雨渡江，岂不快哉。被香山居士，约林和靖，与东坡老，驾勒吾回。'],
    ['uid' => 1007, 'name' => '孙七', 'gender' => '男', 'birth' => 2007, 'text' => '何处相逢？登宝钗楼，访铜雀台。唤厨人斫就，东溟鲸脍；圉人呈罢，西极龙媒。'],
    ['uid' => 1008, 'name' => '周八', 'gender' => '女', 'birth' => 2008, 'text' => '花亦无知，月亦无聊，酒亦无灵。把夭桃斫断，煞他风景；鹦哥煮熟，佐我杯羹。'],
    ['uid' => 1009, 'name' => '吴九', 'gender' => '男', 'birth' => 2009, 'text' => '宫烛分烟，禁池开钥，凤城暮春。向落花香里，澄波影外，笙歌迟日，罗绮芳尘。'],
    ['uid' => 1010, 'name' => '郑十', 'gender' => '女', 'birth' => 2010, 'text' => '谁使神州，百年陆沉，青毡未还？怅晨星残月，北州豪杰；西风斜日，东帝江山。'],
];
foreach ($datas as $data) {
    $params = [
        'index' => $index,
        'id' => $data['uid'],
        'body' => [
            'uid' => $data['uid'],
            'name' => $data['name'],
            'gender' => $data['gender'],
            'birth' => $data['birth'],
            'text' => $data['text'],
        ]
    ];
    try {
        $response = $client->index($params); // 插入和更新都是用index()方法，如果记录不存在则插入，否则更新记录
    } catch (ClientResponseException|MissingParameterException|ServerResponseException $e) {
        $response = $e->getResponse();
    }
    $statusCode = $response->getStatusCode();
    $reasonPhrase = $response->getReasonPhrase();
    if ($statusCode === 201 && $reasonPhrase === 'Created') {
        echo "插入记录[$index][{$data['uid']}]成功" . PHP_EOL;
    } elseif ($statusCode === 200 && $reasonPhrase === 'OK') {
        echo "更新记录[$index][{$data['uid']}]成功" . PHP_EOL;
    } else {
        echo "插入(或更新)记录[$index][{$data['uid']}]失败：$statusCode $reasonPhrase" . PHP_EOL;
    }
}


echo PHP_EOL;
//========== 立即刷新索引 ==========//
try {
    $response = $client->indices()->refresh(['index' => $index]);
} catch (ClientResponseException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $array = $response->asArray();
    $successful = (int)($array['_shards']['successful'] ?? 0);
    $failed = (int)($array['_shards']['failed'] ?? 1);
    $ok = $successful === 1 && $failed === 0;
    if ($ok) {
        echo "刷新索引[$index]成功" . PHP_EOL;
    } else {
        echo "刷新索引[$index]异常：$statusCode $reasonPhrase" . PHP_EOL;
    }
} else {
    echo "刷新索引[$index]失败：$statusCode $reasonPhrase" . PHP_EOL;
}


echo PHP_EOL;
//========== 查询所有记录 ==========//
$params = [
    'index' => $index,
    'size' => 9999, // size为返回记录数，相当于MySQL的LIMIT
    'body' => [],
];
try {
    $response = $client->search($params);
} catch (ClientResponseException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $json = $response->asString();
    $array = json_de($json);
    $rows = $array['hits']['hits'];
    echo '---------- 查询所有记录 ----------' . PHP_EOL;
    foreach ($rows as $row) {
        $id = $row['_id'];
        $name = $row['_source']['name'];
        $gender = $row['_source']['gender'];
        $birth = $row['_source']['birth'];
        $text = $row['_source']['text'];
        echo "e.g.1 [$index][$id] ---> 俺叫{$name}（{$gender}），出生于{$birth}年。【{$text}】" . PHP_EOL;
    }
} else {
    echo "查询索引[$index]所有记录失败：$statusCode $reasonPhrase" . PHP_EOL;
}


echo PHP_EOL;
//========== 查询指定ID记录 ==========//
$params = [
    'index' => $index,
    'size' => 9999, // size为返回记录数，相当于MySQL的LIMIT
    'body' => [
        'query' => [
            'terms' => [
                '_id' => [1003, 1004, 1005],
            ],
        ],
    ]
];
try {
    $response = $client->search($params);
} catch (ClientResponseException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $json = $response->asString();
    $array = json_de($json);
    $rows = $array['hits']['hits'];
    echo '---------- 查询指定ID记录 ----------' . PHP_EOL;
    foreach ($rows as $row) {
        $id = $row['_id'];
        $name = $row['_source']['name'];
        $gender = $row['_source']['gender'];
        $birth = $row['_source']['birth'];
        $text = $row['_source']['text'];
        echo "e.g.2 [$index][$id] ---> 俺叫{$name}（{$gender}），出生于{$birth}年。【{$text}】" . PHP_EOL;
    }
} else {
    echo "查询索引[$index]指定ID记录失败：$statusCode $reasonPhrase" . PHP_EOL;
}


echo PHP_EOL;
//========== 查询包含指定关键字(单个)的记录 ==========//
$params = [
    'index' => $index,
    'size' => 9999, // size为返回记录数，相当于MySQL的LIMIT
    'body' => [
        'query' => [
            'match' => [
                'text' => '风', // 查询text字段中包含“风”的记录
            ],
        ],
    ]
];
try {
    $response = $client->search($params);
} catch (ClientResponseException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $json = $response->asString();
    $array = json_de($json);
    $rows = $array['hits']['hits'];
    echo '---------- 查询包含指定关键字[风]的记录 ----------' . PHP_EOL;
    foreach ($rows as $row) {
        $id = $row['_id'];
        $name = $row['_source']['name'];
        $gender = $row['_source']['gender'];
        $birth = $row['_source']['birth'];
        $text = $row['_source']['text'];
        echo "e.g.3 [$index][$id] ---> 俺叫{$name}（{$gender}），出生于{$birth}年。【{$text}】" . PHP_EOL;
    }
} else {
    echo "查询索引[$index]包含指定关键字(单个)记录失败：$statusCode $reasonPhrase" . PHP_EOL;
}


echo PHP_EOL;
//========== 查询包含指定关键字(多个，或逻辑)的记录 ==========//
$params = [
    'index' => $index,
    'size' => 9999, // size为返回记录数，相当于MySQL的LIMIT
    'body' => [
        'query' => [
            'bool' => [
                'should' => array_map(static function ($keyword) {
                    return ['match' => ['text' => $keyword]];
                }, ['风', '月']),
                'minimum_should_match' => 1
            ]
        ]
    ]
];
try {
    $response = $client->search($params);
} catch (ClientResponseException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $json = $response->asString();
    $array = json_de($json);
    $rows = $array['hits']['hits'];
    echo '---------- 查询包含指定关键字[风]或[月]的记录 ----------' . PHP_EOL;
    foreach ($rows as $row) {
        $id = $row['_id'];
        $name = $row['_source']['name'];
        $gender = $row['_source']['gender'];
        $birth = $row['_source']['birth'];
        $text = $row['_source']['text'];
        echo "e.g.4 [$index][$id] ---> 俺叫{$name}（{$gender}），出生于{$birth}年。【{$text}】" . PHP_EOL;
    }
} else {
    echo "查询索引[$index]包含指定关键字(多个，或逻辑)记录失败：$statusCode $reasonPhrase" . PHP_EOL;
}


echo PHP_EOL;
//========== 查询包含指定关键字(多个，与逻辑)的记录 ==========//
$params = [
    'index' => $index,
    'body' => [
        'query' => [
            'bool' => [
                'must' => array_map(static function ($keyword) {
                    return ['match' => ['text' => $keyword]];
                }, ['风', '月'])
            ]
        ]
    ]
];
try {
    $response = $client->search($params);
} catch (ClientResponseException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $json = $response->asString();
    $array = json_de($json);
    $rows = $array['hits']['hits'];
    echo '---------- 查询包含指定关键字[风]和[月]的记录 ----------' . PHP_EOL;
    foreach ($rows as $row) {
        $id = $row['_id'];
        $name = $row['_source']['name'];
        $gender = $row['_source']['gender'];
        $birth = $row['_source']['birth'];
        $text = $row['_source']['text'];
        echo "e.g.5 [$index][$id] ---> 俺叫{$name}（{$gender}），出生于{$birth}年。【{$text}】" . PHP_EOL;
    }
} else {
    echo "查询索引[$index]包含指定关键字(多个，与逻辑)记录失败：$statusCode $reasonPhrase" . PHP_EOL;
}


echo PHP_EOL;
//========== 删除指定(单个)索引 ==========//
// $deleteIndex = 'my-index-n'; // 指定要删除的索引（该索引下的记录也会被全部删除）
$deleteIndex = $index; // 指定要删除的索引（该索引下的记录也会被全部删除）
try {
    $response = $client->indices()->delete(['index' => $deleteIndex]);
} catch (ClientResponseException|MissingParameterException|ServerResponseException $e) {
    $response = $e->getResponse();
}
$statusCode = $response->getStatusCode();
$reasonPhrase = $response->getReasonPhrase();
if ($statusCode === 200 && $reasonPhrase === 'OK') {
    $array = $response->asArray();
    $ok = isset($array['acknowledged']) && $array['acknowledged'];
    echo "删除索引[$deleteIndex]" . ($ok ? '成功' : '失败') . PHP_EOL;
} else {
    echo "删除索引[$deleteIndex]失败：$statusCode $reasonPhrase" . PHP_EOL;
}


//========== 总结 ==========//
// 1、无论是索引还是记录，其写操作都是异步的，所以执行完写操作后立即读可能得不到期望效果，可先用refresh()方法刷新后再读。
// 2、插入和更新记录都是用index()方法，且无需预先创建索引（如未创建会自动创建），如果记录不存在则插入，否则更新记录。
// 3、API返回的状态码和HTTP状态码在意思表达上基本是一致的，例如返回404表示索引不存在。


//========== 参考资料 ==========//
// 文档：https://www.elastic.co/guide/en/elasticsearch/reference/current/rest-apis.html

Copyright © 2024 码农人生. All Rights Reserved