一、需求分析

注意: 本環境使用 elasticsearch 7.0版本開發，切勿低于此版本

mysql 表結構

有一張表，記錄的數據特別的多，需要將7天前的記錄，插入到Elasticsearch中，并刪除原有表7天前的記錄。

表結構如下：

CREATE TABLE `historic_records` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`user_id` varchar(50) NOT NULL DEFAULT '' COMMENT '用戶id',`time` bigint(20) NOT NULL DEFAULT '0' COMMENT '上線/下線時間',`create_time` bigint(20) NOT NULL DEFAULT '0' COMMENT '創建時間',`update_time` bigint(20) NOT NULL DEFAULT '0' COMMENT '更新時間',`online_status` tinyint(1) NOT NULL DEFAULT '0' COMMENT '在線狀態 默認1 0 離線 1 在線',`status` tinyint(1) NOT NULL DEFAULT '1' COMMENT '軟刪除標志:0-已刪除;1-正常',PRIMARY KEY (`id`),KEY `user_id` (`user_id`),KEY `order_index` (`time`,`create_time`,`update_time`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='歷史記錄表'; View Code

?

查詢sql：

select * from historic_records where create_time < unix_timestamp(date_add(cast(sysdate() as date), interval -7 day)) * 1000

?

刪除sql:

delete from historic_records where create_time < unix_timestamp(date_add(cast(sysdate() as date), interval -7 day)) * 1000

?

ES中的一些概念

index（索引）

相當于mysql中的數據庫

type（類型）

相當于mysql中的一張表

document（文檔）

相當于mysql中的一行（一條記錄）

field（域）

相當于mysql中的一列（一個字段）

節點

一個服務器，由一個名字來標識

集群

一個或多個節點組織在一起

分片

將一份數據劃分為多小份的能力，允許水平分割和擴展容量。多個分片可以響應請求，提高性能和吞吐量。

副本

復制數據，一個節點出問題時，其余節點可以頂上。

倒排索引

可參考 https://www.elastic.co/guide/cn/elasticsearch/guide/current/inverted-index.html

?

es數據結構

設定映射，規定好各個字段及其數據類型，便于es更好地進行管理。根據mysql表結構，映射如下：

# 創建映射 _index_mappings = {"settings": {"index": {"number_of_shards": 3,"number_of_replicas": 1}},"mappings": {# self.index_type : {"properties": {"id": {"type": "long"},"loid": {"type": "keyword"},"mac": {"type": "keyword"},"time": {"type": "date","format": "epoch_millis"},"create_time": {"type": "date","format": "epoch_millis"},"update_time": {"type": "date","format": "epoch_millis"},"online_status": {"type": "short"},"status": {"type": "short"}}# } } } View Code

?

解釋：

索引設置，都在 settings{...} 中

number_of_shards
每個索引的主分片數，默認值是 5 。這個配置在索引創建后不能修改。

number_of_replicas
每個主分片的副本數，默認值是 1 。對于活動的索引庫，這個配置可以隨時修改。

?

映射配置，都在mappings{...} 中

屬性設置，都在?properties{...} 中

?

Elasticsearch 支持?如下簡單域類型：

• 字符串:?string
• 整數 :?byte,?short,?integer,?long
• 浮點數:?float,?double
• 布爾型:?boolean
• 日期:?date

?

仔細看上面的mysql 表結構

由于 id 的類型是?bigint(20)，那么在es中就是?long，表示長整形。

user_id 的類型是?varchar(50) ，在es中，有2中，分別是 text和?keyword。

這2種，是有區別的。text 會創建全文索引，支持模糊搜索。而keyword則不會，必須精確搜索才行。

由于 user_id不需要模糊搜索，因此 設置?keyword才是合理的。

?

create_time 雖然類型是?bigint(20)，但是它存儲在mysql里面，表示時間戳。

因此es中就是data，時間格式為：epoch_millis，表示微秒時間戳。

?

online_status 的類型是tinyint(1)，在es中是?short，表示短的數字

?

三、elasticsearch和kibana搭建

elasticsearch

新建目錄elasticsearch

mkdir /opt/elasticsearch-7.1.1

目錄結構如下：

./ ├── dockerfile ├── elasticsearch-7.1.1-amd64.deb ├── run.sh └── sources.list

?

dockerfile

FROM ubuntu:16.04 # 修改更新源為阿里云 ADD sources.list /etc/apt/sources.list ADD elasticsearch-7.1.1-amd64.deb ./ # 安裝jdk和elasticsearch RUN apt-get update && apt-get install -y openjdk-8-jdk --allow-unauthenticated && apt-get clean all && dpkg -i elasticsearch-7.1.1-amd64.deb && rm -rf elasticsearch-7.1.1-amd64.deb EXPOSE 9200 # 添加啟動腳本 ADD run.sh . RUN chmod 755 run.sh ENTRYPOINT [ "/run.sh"] View Code

?

run.sh

#!/bin/bash set -e# 添加時區 TZ=Asia/Shanghai ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone# 覆蓋配置文件 cp /etc/elasticsearch/elasticsearch.yml /etc/elasticsearch/elasticsearch.yml.bak echo "transport.host: localhost transport.tcp.port: 9300 http.port: 9200 network.host: 0.0.0.0" >> /etc/elasticsearch/elasticsearch.yml # 修改啟動文件,去掉-d參數,避免后臺運行 sed -i 72's@-d -p $PID_FILE@-p $PID_FILE@g' /etc/init.d/elasticsearch# 啟動elasticsearch,要hold住,否則容器啟動就退出了! /etc/init.d/elasticsearch start View Code

?

sources.list

deb http://mirrors.aliyun.com/ubuntu/ xenial main restricted deb http://mirrors.aliyun.com/ubuntu/ xenial-updates main restricted deb http://mirrors.aliyun.com/ubuntu/ xenial universe deb http://mirrors.aliyun.com/ubuntu/ xenial-updates universe deb http://mirrors.aliyun.com/ubuntu/ xenial multiverse deb http://mirrors.aliyun.com/ubuntu/ xenial-updates multiverse deb http://mirrors.aliyun.com/ubuntu/ xenial-backports main restricted universe multiverse deb http://mirrors.aliyun.com/ubuntu xenial-security main restricted deb http://mirrors.aliyun.com/ubuntu xenial-security universe deb http://mirrors.aliyun.com/ubuntu xenial-security multiverse View Code

?

生成鏡像

docker build -t elasticsearch-7.1.1 .

?

啟動容器

docker run -d -it --restart=always -p 9200:9200 elasticsearch-7.1.1

?

訪問頁面

?

?

kibana

新建目錄kibana

mkdir /opt/kibana-7.1.1

目錄結構如下：

./ ├── dockerfile ├── kibana-7.1.1-amd64.deb └── run.sh

?

dockerfile

FROM ubuntu:16.04 ADD kibana-7.1.1-amd64.deb ./ # 安裝jdk和elasticsearch RUN dpkg -i kibana-7.1.1-amd64.deb && rm -rf kibana-7.1.1-amd64.deb EXPOSE 5601 # 添加啟動腳本 ADD run.sh . RUN chmod 755 run.sh ENTRYPOINT [ "/run.sh"] View Code

?

run.sh

#!/bin/bash# 添加時區 TZ=Asia/Shanghai ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone#elasticsearch="192.168.91.128" if [ -z $elasticsearch ];thenecho "elasticsearch參數為空!比如: 192.168.91.128"exit fi# 修改配置文件 # 修改監聽地址 sed -i '7s@#server.host: "localhost"@server.host: "0.0.0.0"@g' /etc/kibana/kibana.yml # 刪除行，并添加一行內容 sed -i '28d' /etc/kibana/kibana.yml sed -i "N;28 i elasticsearch.hosts: ["http://$elasticsearch:9200"]" /etc/kibana/kibana.yml # 啟動 /usr/share/kibana/bin/kibana "-c /etc/kibana/kibana.yml" View Code

?

生成鏡像

docker build -t kibana-7.1.1 .

?

啟動鏡像

docker run -d -it --restart=always -p 5601:5601 -e elasticsearch=192.168.10.104 kibana-7.1.1

?

訪問頁面

?

二、查詢mysql數據

為了方便操作 mysql，封裝了一個mysql工具類，用來查詢和更新數據。

mysql.py

#!/usr/bin/env python3 # coding: utf-8import pymysqlclass Mysql(object):# mysql 端口號,注意：必須是int類型def __init__(self,host,user,passwd,db_name,port=3306):self.host = hostself.user = userself.passwd = passwdself.db_name = db_nameself.port = portdef select(self,sql):"""執行sql命令:param sql: 命令:return: 元祖"""try:conn = pymysql.connect(host=self.host,user=self.user,passwd=self.passwd,port=self.port,database=self.db_name,charset='utf8',cursorclass=pymysql.cursors.DictCursor)cur = conn.cursor() # 創建游標cur.execute(sql) # 執行sql命令res = cur.fetchall() # 獲取執行的返回結果 cur.close()conn.close() # 關閉mysql 連接return resexcept Exception as e:print(e)return Falsedef update(self,sql):"""更新操作，比如insert, delete,update:param sql: sql命令:return: bool"""try:conn = pymysql.connect(host=self.host,user=self.user,passwd=self.passwd,port=self.port,database=self.db_name,)cur = conn.cursor(cursor=pymysql.cursors.DictCursor) # 創建游標# conn.cursor()# print("ip: {} insert 執行命令: {}".format(self.host,sql))sta = cur.execute(sql) # 執行sql命令，返回影響的行數# print("sta",sta,type(sta))#res = cur.fetchall() # 獲取執行的返回結果if isinstance(sta,int): # 判斷返回結果, 是數字就是正常的#print('插入記錄 Done')pass# write_log('正常，遠程執行sql: %s 成功'%sql, "green")else:write_log('錯誤，遠程執行sql: %s 失敗'%sql, "red")return Falseconn.commit() # 主動提交，否則執行sql不生效 cur.close()conn.close() # 關閉mysql 連接return staexcept Exception as e:print(e)# write_log('錯誤，遠程mysql執行命令: {} 異常'.format(sql), "red")return False View Code

?

使用時，就簡單了。導入這個類，調用相關方法。

mysql_test.py

from mysql import Mysqlhost = "192.168.0.179" user = "sdn_db" passwd = "Sdn@ujmyhn" db_name = "terminalservice" port = 3306sql = "select * from terminals_record_0 where create_time < unix_timestamp(date_add(cast(sysdate() as date), interval -7 day)) * 1000" res = Mysql(host,user,passwd,db_name,port).select(sql) print(res) View Code

?

?

三、完整代碼

由于時間關系，代碼不一一解釋了。附上完整代碼：

./ ├── conf.py ├── es_bulk.py ├── README.md ├── requirements.txt └── utils├── common.py└── mysql.py

?

conf.py

#!/usr/bin/env python3 # coding: utf-8 """ 配置文件，用于mysql和elasticsearch """ import os BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # 項目根目錄# mysql HOST = "192.168.0.136" USER = "root" PASSWD = "123456" DB_NAME = "terminal" PORT = 3306# elasticsearch INDEX_NAME = "historic_records" INDEX_TYPE = "_doc" ES_IP = "192.169.3.133"MAXIMUM = 100 # 一次性插入多少條 View Code

?

es_bulk.py

#!/usr/bin/env python3 # coding: utf-8import time from elasticsearch import Elasticsearch from elasticsearch import helpersimport conf from utils.mysql import Mysql from utils.common import write_log,valid_ip,check_tcpclass ElasticObj:def __init__(self,timeout=3600):''':param timeout: 超時時間'''self.index_name = conf.INDEX_NAME # 索引名稱self.index_type = conf.INDEX_TYPE # 索引類型self.es_ip = conf.ES_IP # es ip# 無用戶名密碼狀態self.es = Elasticsearch([self.es_ip], port=9200, timeout=timeout)# 用戶名密碼狀態# self.es = Elasticsearch([self.es_ip], http_auth=('esadm', 'mdase123'), port=9200, timeout=timeout)def create_index(self):'''創建索引:return: bool'''# 創建映射_index_mappings = {# 索引配置"settings": {"index": {"number_of_shards": 3, # 分片數"number_of_replicas": 1 # 副本數 }},# 設置字段"mappings": {"properties": {"id": {"type": "long"},"loid": {"type": "keyword"},"mac": {"type": "keyword"},"time": {"type": "date","format": "epoch_millis"},"create_time": {"type": "date","format": "epoch_millis"},"update_time": {"type": "date","format": "epoch_millis"},"online_status": {"type": "short"},"status": {"type": "short"}}}}# 判斷索引不存在時if self.es.indices.exists(index=self.index_name) is not True:# 創建索引res = self.es.indices.create(index=self.index_name, body=_index_mappings)# print(res)if not res:write_log("錯誤，創建索引{}失敗".format(self.index_name),"red")return Falsewrite_log("正常，創建索引{}成功".format(self.index_name), "green")return Trueelse:write_log("正常，索引{}已存在".format(self.index_name), "green")return Truedef bulk_insert(self,table,data_list):"""批量寫入數據:param table: 表名:param data_list: 數據列表[{'online_status': 1,'update_time': 1556073035327,'create_time': 1556073035327,'id': 1, 'status': 1,'time': 1556073035327,'loid': '100010000123','mac': '60:45:cb:87:c9:93'},...]:return: bool"""# 批量插入start_time = time.time() # 開始時間actions = [] # 臨時數據列表i = 0 # 計數值try:# 循環數據列表for data in data_list:action = {"_index": self.index_name,"_type": self.index_type,#"_id": i, #_id 也可以默認生成，不賦值"_source": {'id': data['id'],'user_id': data['user_id'],'time': data['time'],'create_time': data['create_time'],'online_status': data['online_status'],'status': data['status'],}}i += 1actions.append(action) # 添加到列表if len(action) == conf.MAXIMUM: # 列表數量達到100時helpers.bulk(self.es, actions) # 批量插入數據del actions[0:len(action)] # 刪除列表元素if i > 0: # 不足100時,插入剩余數據 helpers.bulk(self.es, actions)end_time = time.time() # 結束時間t = round((end_time - start_time),2) # 計算耗時# print('本次共寫入{}條數據，用時{}s'.format(i, t))write_log("正常，{} 表寫入ES {}條數據，用時{}s".format(table,i, t), "green")return Trueexcept Exception as e:print(e)return Falsedef has_table(self,db_name,target_table):"""遠程表是否存在:return: bool"""mysql_obj = Mysql(conf.HOST, conf.USER, conf.PASSWD, conf.DB_NAME, conf.PORT)sql = "select count(1) from {}.{}".format(db_name, target_table)res = mysql_obj.select(sql)# print("表是否存在",res,type(res))if res is False:write_log("錯誤，遠程表 {}.{} 不存在".format(db_name,target_table),"red")return Falseelse:return Truedef has_conf(self):"""判斷配置文件中的mysql和es 端口是否正常:return:"""if not valid_ip(conf.HOST):write_log("錯誤，MySQL IP配置不正確","red")return Falseif not valid_ip(conf.ES_IP):write_log("錯誤，ES IP配置不正確","red")return Falseif not check_tcp(conf.HOST,conf.PORT):write_log("錯誤，MySQL {} 端口不可達".format(conf.PORT),"red")return Falseif not check_tcp(conf.ES_IP,9200):write_log("錯誤，ES 9200 端口不可達","red")return Falsereturn Truedef read_mysql_es(self):"""讀取7天的記錄，并寫入es:return: bool"""# 判斷配置文件中的mysql和es 端口是否正常if not self.has_conf():# print(1)return False# 創建索引if self.create_index() is False:# print(2)return Falsemax = conf.MAXIMUM # 一次性查詢多少條 flag_list = [] # 標志位列表mysql_obj = Mysql(conf.HOST, conf.USER, conf.PASSWD, conf.DB_NAME, conf.PORT)for i in range(64): # 寫入64張表# 判斷表是否存在res = self.has_table(conf.DB_NAME,'historic_record_%s'%i)if not res:flag_list.append(False)return Falseid = 0 # 每一次查詢后的最大idwhile True:# 查詢數據sql = "select * from historic_record_%s where create_time < unix_timestamp(date_add(cast(sysdate() as date), interval -7 day)) * 1000 and id > %s order by id limit %s" % (i, id, max)# print(sql)data_list = mysql_obj.select(sql)# print(data_list)if not data_list: # 當結果為空時,結束循環write_log("警告，執行sql: %s 記錄為空，無需寫入es" %(sql), "yellow")break # 跳出循環 last_row = data_list[-1] # 最后一行記錄# print(last_row)id = last_row['id'] # 修改最大id res = self.bulk_insert('historic_record_%s' % i, data_list)if not res:write_log("錯誤，historic_record_%s 寫入ES 失敗"%i,"red")flag_list.append(False)return Falseif False in flag_list:write_log("錯誤，historic_record 部分表寫入ES錯誤，請查看上文","red")return Falsewrite_log("正常，historic_record 64張表全部寫入ES成功", "green")return Truedef delete_record(self):"""刪除7天的表數據:return: bool"""max = conf.MAXIMUM # 一次性查詢多少條flag_list = []mysql_obj = Mysql(conf.HOST, conf.USER, conf.PASSWD, conf.DB_NAME, conf.PORT)for i in range(64): # 64張表# 判斷表是否存在res = self.has_table(conf.DB_NAME, 'historic_record_%s' % i)if not res:flag_list.append(False)return False### 先查詢數據id = 0 # 每一次查詢后的最大idwhile True:# 查詢數據sql = "select * from historic_record_%s where create_time < unix_timestamp(date_add(cast(sysdate() as date), interval -7 day)) * 1000 and id > %s order by id limit %s" % (i, id, max)# print(sql)data_list = mysql_obj.select(sql)# print(data_list)if not data_list: # 當結果為空時,結束循環write_log("警告，執行sql: %s 記錄為空，無需刪除" % sql, "yellow")break # 跳出循環### 再刪除數據sql = "delete from historic_record_%s where create_time < unix_timestamp(date_add(cast(sysdate() as date), interval -7 day)) * 1000 and id > %s order by id limit %s" % (i, id, max)# print(sql)res = mysql_obj.update(sql)if res is False:write_log("錯誤，刪除 historic_record_%s 記錄失敗" % i, "red")flag_list.append(False)breakelse:write_log("正常，刪除 historic_record_%s 記錄成功" % i, "green")last_row = data_list[-1] # 最后一行記錄# print(last_row)id = last_row['id'] # 修改最大idif False in flag_list:write_log("錯誤，刪除 historic_record 部分表失敗，請查看上文", "red")return Falsewrite_log("正常，刪除 historic_record 64張表記錄全部成功", "green")def main(self):self.read_mysql_es()self.delete_record()ElasticObj().main() # 執行主程序 View Code

?

common.py

#!/usr/bin/env python3 # coding: utf-8 """ 共有的方法 """import sys import iodef setup_io(): # 設置默認屏幕輸出為utf-8編碼sys.stdout = sys.__stdout__ = io.TextIOWrapper(sys.stdout.detach(), encoding='utf-8', line_buffering=True)sys.stderr = sys.__stderr__ = io.TextIOWrapper(sys.stderr.detach(), encoding='utf-8', line_buffering=True) setup_io()import os import time import conf import socket import subprocess import ipaddress from multiprocessing import cpu_countdef write_log(content,colour='white',skip=False):"""寫入日志文件:param content: 寫入內容:param colour: 顏色:param skip: 是否跳過打印時間:return:"""# 顏色代碼colour_dict = {'red': 31, # 紅色'green': 32, # 綠色'yellow': 33, # 黃色'blue': 34, # 藍色'purple_red': 35, # 紫紅色'bluish_blue': 36, # 淺藍色'white': 37, # 白色 }choice = colour_dict.get(colour) # 選擇顏色 path = os.path.join(conf.BASE_DIR,"output.log") # 日志文件with open(path, mode='a+', encoding='utf-8') as f:if skip is False: # 不跳過打印時間時content = time.strftime('%Y-%m-%d %H:%M:%S') + ' ' + contentinfo = "\033[1;{};1m{}\033[0m".format(choice, content)print(info)f.write(content+"\n")def execute_linux2(cmd, timeout=10, skip=False):"""執行linux命令,返回list:param cmd: linux命令:param timeout: 超時時間,生產環境, 特別卡, 因此要3秒:param skip: 是否跳過超時:return: list"""p = subprocess.Popen(cmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, shell=True)# print(p)# timeout = 1 # 超時時間t_beginning = time.time() # 開始時間# seconds_passed = 0 # 執行時間while True:if p.poll() is not None:breakseconds_passed = time.time() - t_beginningif timeout and seconds_passed > timeout:p.terminate()# raise TimeoutError(cmd, timeout)if not skip:# self.res.code = 500# print('命令: {},執行超時!'.format(cmd))write_log('錯誤, 命令: {},本地執行超時!'.format(cmd),"red")# return self.res.__dict__return False# return '命令: {},執行超時!'.format(cmd)# result = p.stdout.read().decode('utf-8').strip() # 命令運行結果# print("result",result)# self.res.data = result# return self.res.__dict__result = p.stdout.readlines()return resultdef valid_ip(ip):"""驗證ip是否有效,比如192.168.1.256是一個不存在的ip:return: bool"""try:# 判斷 python 版本if sys.version_info[0] == 2:ipaddress.ip_address(ip.strip().decode("utf-8"))elif sys.version_info[0] == 3:# ipaddress.ip_address(bytes(ip.strip().encode("utf-8"))) ipaddress.ip_address(ip)return Trueexcept Exception as e:print(e)return Falsedef check_tcp(ip, port, timeout=1):"""檢測tcp端口:param ip: ip地址:param port: 端口號:param timeout: 超時時間:return: bool"""flag = Falsetry:socket.setdefaulttimeout(timeout) # 整個socket層設置超時時間cs = socket.socket(socket.AF_INET, socket.SOCK_STREAM)address = (str(ip), int(port))status = cs.connect_ex((address)) # 開始連接 cs.settimeout(timeout)if not status:flag = Truereturn flagexcept Exception as e:print(e)return flagCOROUTINE_NUMBER = cpu_count() # 協程池數量，根據cpu核心數來開，避免cpu飆高 View Code

?

mysql.py

#!/usr/bin/env python3 # coding: utf-8import pymysql from utils.common import write_logclass Mysql(object):# mysql 端口號,注意：必須是int類型def __init__(self,host,user,passwd,db_name,port=3306):self.host = hostself.user = userself.passwd = passwdself.db_name = db_nameself.port = portdef select(self,sql):"""執行sql命令:param sql: 命令:return: 元祖"""try:# print(host,self.user,self.passwd,self.port,self.db_name)conn = pymysql.connect(host=self.host,user=self.user,passwd=self.passwd,port=self.port,database=self.db_name,charset='utf8',cursorclass=pymysql.cursors.DictCursor)cur = conn.cursor() # 創建游標# conn.cursor()cur.execute(sql) # 執行sql命令res = cur.fetchall() # 獲取執行的返回結果 cur.close()conn.close() # 關閉mysql 連接return resexcept Exception as e:print(e)return Falsedef update(self,sql):"""更新操作，比如insert, delete,update:param sql: sql命令:return: bool"""try:conn = pymysql.connect(host=self.host,user=self.user,passwd=self.passwd,port=self.port,database=self.db_name,)cur = conn.cursor(cursor=pymysql.cursors.DictCursor) # 創建游標# conn.cursor()# print("ip: {} insert 執行命令: {}".format(self.host,sql))sta = cur.execute(sql) # 執行sql命令，返回影響的行數# print("sta",sta,type(sta))#res = cur.fetchall() # 獲取執行的返回結果if isinstance(sta,int): # 判斷返回結果, 是數字就是正常的#print('插入記錄 Done')pass# write_log('正常，遠程執行sql: %s 成功'%sql, "green")else:write_log('錯誤，遠程執行sql: %s 失敗'%sql, "red")return Falseconn.commit() # 主動提交，否則執行sql不生效 cur.close()conn.close() # 關閉mysql 連接#Migration.flag_list.append(True)return staexcept Exception as e:print(e)# write_log('錯誤，遠程mysql執行命令: {} 異常'.format(sql), "red")# Migration.flag_list.append(False)return False View Code

?

requirements.txt

PyMySQL==0.9.2 elasticsearch==6.3.1

?

README.md

## 說明 終端歷史記錄表，寫入到elasticsearch中。主要將(terminal.historic_record_0~63) 這64張表的7天前數據寫入到elasticsearch中并刪除 64張表的7天前記錄`注意: 本環境使用 elasticsearch 7.0版本開發，切勿低于此版本`## 配置說明 `conf.py` 是環境配置主要修改 以下信息 ```python # mysql HOST = "192.168.0.136" USER = "root" PASSWD = "123456" DB_NAME = "terminal" PORT = 3306# elasticsearch INDEX_NAME = "historic_record" INDEX_TYPE = "_doc" ES_IP = "192.169.3.133" ```請根據實際情況修改以上變量## 運行說明 ## 一鍵執行，遷移相關所有表 `python es_bulk.py`## 查看結果 結果會輸出到`output.log`文件，直接查看即可!登錄到`kibana`，查看數據是否存在<br/> <br/> Copyright (c) 2019-present, xiao You View Code

?

注意：如果是es 6.x的版本，創建索引，需要增加?index_type，否則會報錯。

比如：

# 創建映射 _index_mappings = {# 索引配置"settings": {"index": {"number_of_shards": 3, # 分片數"number_of_replicas": 1 # 副本數 }},# 設置字段"mappings": {self.index_type: {"properties": {"id": {"type": "long"},"loid": {"type": "keyword"},"mac": {"type": "keyword"},"time": {"type": "date","format": "epoch_millis"},"create_time": {"type": "date","format": "epoch_millis"},"update_time": {"type": "date","format": "epoch_millis"},"online_status": {"type": "short"},"status": {"type": "short"}}}} }

?

?

本文參考鏈接：

https://www.cnblogs.com/aaanthony/p/7380662.html

https://blog.csdn.net/m0_37673307/article/details/81153700

?

轉載于:https://www.cnblogs.com/xiao987334176/p/10885295.html

總結

以上是生活随笔為你收集整理的python MySQL 插入Elasticsearch的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。