1、將數(shù)據(jù)寫入到mysql中

• 創(chuàng)建測試表

CREATE TABLE `student` (
  `sno` varchar(10) COLLATE utf8mb4_unicode_ci NOT NULL,
  `sname` varchar(20) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
  `sage` int(2) DEFAULT NULL,
  `ssex` varchar(5) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
  `description` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT NULL,
  PRIMARY KEY (`sno`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

• 測試表插入數(shù)據(jù)

insert into student values ('s001','張三',23,'男','張三是個好學(xué)生'); 
insert into student values ('s002','李四',23,'男','張三是個好學(xué)生');  
insert into student values ('s003','吳鵬',25,'男','張三是個好學(xué)生');  
insert into student values ('s004','琴沁',20,'女','張三是個好學(xué)生');  
insert into student values ('s005','王麗',20,'女','張三是個好學(xué)生');  
insert into student values ('s006','李波',21,'男','張三是個好學(xué)生');  
insert into student values ('s007','劉玉',21,'男','張三是個好學(xué)生');  
insert into student values ('s008','蕭蓉',21,'女','張三是個好學(xué)生');  
insert into student values ('s009','陳蕭曉',23,'女','張三是個好學(xué)生');  
insert into student values ('s010','陳美',22,'女','張三是個好學(xué)生');

• 表字段描述

• 字段意義

sno 學(xué)號
sname	學(xué)生姓名
sage	學(xué)生年齡
ssex	學(xué)生性別
description	學(xué)生描述

2、python安裝fastapi、elasticsearch框架、Manticore Search框架和mysql客戶端

pip install elasticsearch==8.8.2
pip install pymysql
pip install manticoresearch

# Uvicorn是一個ASGI服務(wù)器，用于運(yùn)行FastAPI應(yīng)用。
pip install uvicorn
pip install fastapi

3、將mysql的數(shù)據(jù)寫入到elasticsearch中

• mysql數(shù)據(jù)同步到es

# mysql-to-es
# 本地es版本 8.8.2
# python es版本  8.8.2
import pymysql
from elasticsearch import Elasticsearch

def get_data():
    conn=pymysql.connect(host="localhost",port=3306,user="root",password="root",database="mydb")
    cursor=conn.cursor()
    sql="select * from student"
    t_header_sql = "desc student "
    cursor.execute(sql)
    results=cursor.fetchall()
    cursor.execute(t_header_sql)
    header_results = cursor.fetchall()
    conn.close()
    return results,header_results

def create_es_data():
    # es = Elasticsearch()
    es = Elasticsearch(
        [
            "https://192.168.10.1:9200",
        ],
        ca_certs="./path/to/http_ca.crt",	# es認(rèn)證證書，8.0版本后開始使用
        basic_auth=("elastic", "U4mRQUoVeQ+YMzcCFD1t"),
        request_timeout=3600
    )
    try:
        data_results,cloumns_results = get_data()
        for row in data_results:
            message = {}
            for i in range(len(row)):
                # print(cloumns_results[i][0], row[i])
                message[cloumns_results[i][0]] = row[i]

            print(message)
            es.index(index="student", document=message)
    except Exception as e:
        print("Error:" + str(e))

if __name__=="__main__":
    create_es_data()

• es查看數(shù)據(jù)（Elasticvue插件）

?文章來源地址http://www.zghlxwxcb.cn/news/detail-650262.html

4、將mysql的數(shù)據(jù)寫入到Manticore中

• mysql數(shù)據(jù)同步到Manticore

注：Manticore 和 Mysql 使用pymysql即mysql客戶端

import pymysql

def get_data():
    conn = pymysql.connect(host="localhost", port=3306, user="root", password="root", database="mydb")
    cursor = conn.cursor()
    sql = "select * from student"
    t_header_sql = "desc student "
    cursor.execute(sql)
    results = cursor.fetchall()
    cursor.execute(t_header_sql)
    header_results = cursor.fetchall()
    header_tuple = tuple(x[0] for x in header_results)
    conn.close()
    return results, header_tuple

def inster_data_to_manticore():
    try:
        db = pymysql.connect(host='localhost', port=9306)
        print('manticoredb 連接成功！')
    except:
        print('something wrong!')

    if db:
        cursor = db.cursor()
        rows, header_tuple = get_data()
        header_str = str(header_tuple).replace("\'","")
        sql = 'drop table if exists students'
        cursor.execute(sql)
        db.commit()

        for row in rows:
            sql = f'INSERT INTO students{header_str} VALUES {row}'
            print(sql)
            cursor.execute(sql)
            db.commit()

if __name__ == '__main__':
    inster_data_to_manticore()

• Manticore 數(shù)據(jù)查詢(工具Webyog SQLyog)

?

5、elasticsearch查找類的封裝

• es安全認(rèn)證連接（參考官網(wǎng)）
• 按fields查詢方法封裝，輸入?yún)?shù)fields 篩選器，query查詢字符串

# elasticsearch_query_class.py
from elasticsearch import Elasticsearch
class elasticsearchself():
    def __init__(self,index_name,index_type = '_doc'):
        self.es = Elasticsearch(
        [
            "https://192.168.10.1:9200",
        ],
        ca_certs="./path/to/http_ca.crt",
        basic_auth=("elastic", "U4mRQUoVeQ+YMzcCFD1t"),
        request_timeout=3600
        )
        self.index_name=index_name
        self.index_type=index_type

    def search_by_fields(self,query,fields,count:int=30):
        ds = {"multi_match": {"query": query,"fields": fields}}
        fields = ['sname','description']
        match_data=self.es.search(index=self.index_name,query=ds,size=count)
        return match_data

es=elasticsearchself(index_name="student")
query = "張三"
fields= ['sname','description']
match_data = es.search_by_fields(query,fields)
print(match_data)

6、Manticoresearch查找類的封裝

# manticoreself.py

import manticoresearch
from manticoresearch.api import search_api
from manticoresearch.model.search_request import SearchRequest

class manticoresearchself():
    def __init__(self, index_name):
        self.configuration = manticoresearch.Configuration(
        host = "http://127.0.0.1:9308"
        )
        self.index_name = index_name

    def search_all_text(self,query):
        with manticoresearch.ApiClient(self.configuration) as api_client:
            # Create an instance of the API class
            api_instance = search_api.SearchApi(api_client)

            # # Create SearchRequest
            # search_request = SearchRequest()
            # search_request.index='students'
            # # search_request.fulltext_filter=QueryFilter(23)
            # search_request.fulltext_filter =

            search_request = SearchRequest(
                index='students',
                query={
                    "match":
                        {
                            "*" : query
                        }
                },
            )

            # example passing only required values which don't have defaults set
            try:
                # Performs a search
                api_response = api_instance.search(search_request)
                # pprint(api_response)
                return api_response
            except manticoresearch.ApiException as e:
                print("Exception when calling SearchApi->search: %s\n" % e)


mc=manticoresearchself(index_name="student")
query = "s004"
match_data = mc.search_all_text(query)
match_data_dict = match_data.__dict__
print(match_data._hits._hits[0]["_source"])

7、fastapi實現(xiàn)elasticseach的全文檢索模糊查詢

# main.py
from fastapi import FastAPI
from elasticsearch_query_class import elasticsearchself
import json
import time

app = FastAPI()

@app.get("/get_es/{query}")
async def get_es(query):
    fields = ['*']
    es=elasticsearchself(index_name="student")
    time_start = time.time()  # 記錄開始時間
    data=es.search_by_fields(query,fields)
    time_end = time.time()  # 記錄結(jié)束時間
    address_data=data["hits"]["hits"]
    address_list=[]
    for item in address_data:
        address_list.append(item["_source"])
    time_sum = time_end - time_start  # 計算的時間差為程序的執(zhí)行時間，單位為秒/s
    address_list.append({"time_sum":time_sum})
    new_json=json.dumps(address_list,ensure_ascii=False)
    return json.loads(new_json)

思路: es創(chuàng)建篩選器列表fields,[*]表示所有字段，查詢體query

8、fastapi實現(xiàn)Manticoresearch的全文檢索篩選查詢

# main.py
from fastapi import FastAPI
from manticoreself import manticoresearchself
import json
import time

app = FastAPI()

@app.get('/get_mc/<query>')
async def get_mc(query):
    mc = manticoresearchself(index_name="student")
    time_start = time.time()  # 記錄開始時間
    data = mc.search_all_text(query)
    time_end = time.time()  # 記錄結(jié)束時間
    address_data = data._hits._hits
    address_list = []
    for item in address_data:
        address_list.append(item["_source"])

    time_sum = time_end - time_start  # 計算的時間差為程序的執(zhí)行時間，單位為秒/s
    address_list.append({"time_sum": time_sum})
    new_json = json.dumps(address_list, ensure_ascii=False)
    return json.loads(new_json)

思路: Manticoresearch 支持openapi查詢接口，使用search_all_text api接口查詢結(jié)果，查詢體query

9、fastapi實現(xiàn)的Mysql列表字段全文查詢

# main.py
from fastapi import FastAPI
import json
import time
import pymysql

app = FastAPI()

@app.get('/get_mysql/<query>')
async def get_mysql(query):
    conn = pymysql.connect(host="localhost", port=3306, user="root", password="root", database="mydb")
    cursor = conn.cursor()
    fields = "(sname, description, sno,ssex)"

    try:
        create_full_index_sql = f"create fulltext index full_idx_to_table on student{fields} "
        cursor.execute(create_full_index_sql)
    except:
        pass

    query_sql = f"select * from student where match{fields} against('{query}')"
    time_start = time.time()  # 記錄開始時間
    cursor.execute(query_sql)
    time_end = time.time()  # 記錄結(jié)束時間
    results = cursor.fetchall()
    address_list = []
    for row in results:
        address_list.append(row)
    time_sum = time_end - time_start  # 計算的時間差為程序的執(zhí)行時間，單位為秒/s
    address_list.append({"time_sum": time_sum})
    new_json = json.dumps(address_list, ensure_ascii=False)
    return json.loads(new_json)

思路: 先創(chuàng)建需要查找的字段即篩選器，利用篩選器列表創(chuàng)建全文檢索index(full_idx_to_table ) 后，使用fts(Full-Text Search)查詢

10、fastapi實現(xiàn)接口源碼即驗證

• fastapi源碼

# main.py

from fastapi import FastAPI
from elasticsearch_query_class import elasticsearchself
import json
from manticoreself import manticoresearchself
import time
import pymysql

app = FastAPI()

@app.get("/get_es/{query}")
async def get_es(query):
    fields = ['*']
    es=elasticsearchself(index_name="student")
    time_start = time.time()  # 記錄開始時間
    data=es.search_by_fields(query,fields)
    time_end = time.time()  # 記錄結(jié)束時間
    address_data=data["hits"]["hits"]
    address_list=[]
    for item in address_data:
        address_list.append(item["_source"])
    time_sum = time_end - time_start  # 計算的時間差為程序的執(zhí)行時間，單位為秒/s
    address_list.append({"time_sum":time_sum})
    new_json=json.dumps(address_list,ensure_ascii=False)
    return json.loads(new_json)

@app.get('/get_mc/<query>')
async def get_mc(query):
    mc = manticoresearchself(index_name="student")
    time_start = time.time()  # 記錄開始時間
    data = mc.search_all_text(query)
    time_end = time.time()  # 記錄結(jié)束時間
    address_data = data._hits._hits
    address_list = []
    for item in address_data:
        address_list.append(item["_source"])

    time_sum = time_end - time_start  # 計算的時間差為程序的執(zhí)行時間，單位為秒/s
    address_list.append({"time_sum": time_sum})
    new_json = json.dumps(address_list, ensure_ascii=False)
    return json.loads(new_json)

@app.get('/get_mysql/<query>')
async def get_mysql(query):
    conn = pymysql.connect(host="localhost", port=3306, user="root", password="root", database="mydb")
    cursor = conn.cursor()
    fields = "(sname, description, sno,ssex)"

    try:
        create_full_index_sql = f"create fulltext index full_idx_to_table on student{fields} "
        cursor.execute(create_full_index_sql)
    except:
        pass

    query_sql = f"select * from student where match{fields} against('{query}')"
    time_start = time.time()  # 記錄開始時間
    cursor.execute(query_sql)
    time_end = time.time()  # 記錄結(jié)束時間
    results = cursor.fetchall()
    address_list = []
    for row in results:
        address_list.append(row)
    time_sum = time_end - time_start  # 計算的時間差為程序的執(zhí)行時間，單位為秒/s
    address_list.append({"time_sum": time_sum})
    new_json = json.dumps(address_list, ensure_ascii=False)
    return json.loads(new_json)

• 驗證(地址：http://127.0.0.1:8000/dcos)

?

?

?

10、總結(jié)

1. elasticsearch 全文搜索

• 支持中、英全文搜索
• 但速度沒有mysql和Manticore Search快

1. Manticore Search全文搜索

• 暫時只支持英文全文搜索
• 搜索速度快，相比es少量數(shù)據(jù)快，沒有mysql少量數(shù)據(jù)快，但據(jù)官方顯示大量數(shù)據(jù)時Manticore Search快于mysql

1. Mysql全文搜索

• 支持中、英搜索，中文時sql ?against中加IN BOOLEAN MODE
• 少量數(shù)據(jù)時搜索極快，但全文搜索時需要創(chuàng)建搜索數(shù)據(jù)的全文索引，有些麻煩

到了這里，關(guān)于fastapi結(jié)合Manticore Search、elasticsearch、mysql實現(xiàn)全文搜索的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！