MySQL DDL转ElasticSearch Mapping(用于在Kibana中创建索引).

sql_to_es_mapping.py

'''

MySQL DDL转ElasticSearch Mapping(用于在Kibana中创建索引).

输出(示例):

新增索引:

PUT /t_comm_demo
{
    "mappings":{
        "properties":{
            "id":{
                "type":"long"
            },
            "comm_id":{
                "type":"long"
            },
            "comm_notice":{
                "type":"integer"
            },
            "comm_picture":{
                "type":"keyword"
            },
            "comm_time":{
                "type":"date"
            },
            "gu_amount":{
                "type":"double"
            },
            "user_sex_content":{
                "type":"text"
            },
            "loan_term":{
                "type":"integer"
            }
        }
    }
}

查看映射:

GET /t_comm_demo/_mapping


------------------------------------------

author: puruidong
version: 2022-03-02

'''

sql='''
CREATE TABLE `t_comm_demo` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT 'id',
  `comm_id` bigint DEFAULT NULL COMMENT '',
  `comm_notice` tinyint(1) DEFAULT NULL COMMENT '',
  `comm_picture` json DEFAULT NULL COMMENT '',
  `comm_time` timestamp NULL DEFAULT NULL COMMENT '',
  `gu_amount` decimal(10,2) DEFAULT NULL COMMENT '',
  `user_sex_content` varchar(8) DEFAULT NULL COMMENT '',
  `loan_term` int DEFAULT NULL COMMENT '',
  PRIMARY KEY (`id`),
) ENGINE=InnoDB AUTO_INCREMENT=126 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='示例表';
'''


import json

# ElasticSearch 关键字列表-自行配置-在此列表中的列名,会使用keyword类型.
keywordList=["comm_picture"]


'''
数据类型映射.

映射规则(部分,如需要可自行完善):

MySQL(8.0) -> ElasticSearch(7.9.3)
-----------------------
varchar -> text
int -> integer
bigint -> long
decimal -> double
tinyint -> integer
json -> nested
timestamp -> date
double -> double

'''
dataTypeMapping={"varchar":"text","int":"integer","bigint":"long","decimal":"double","double":"double","tinyint":"integer","json":"nested","timestamp":"date"}


def trans():
  dataInfoMap={}
  tableName=""
  for item in sql.split("\n"):
    if item.find("CREATE")!=-1:
      tableName=item.split(" ")[2].replace("`","").lower()
      break
  for item in sql.split("\n"):
    if item.find("CREATE")==-1 and item.find("ENGINE")==-1 and item.find("KEY")==-1:
      itemArr=item.split(" ")
      if len(itemArr)>4:
        itemName=itemArr[2].replace("`","")
        esDataType=itemArr[3]
        try:
          if keywordList.index(itemName)!=-1:
            esDataType="keyword"
        except ValueError as e:
          dataType=itemArr[3]
          bracketIndex=itemArr[3].find("(")
          if bracketIndex!=-1:
            dataType=itemArr[3][0:bracketIndex]
          esDataType=dataTypeMapping.get(dataType)
        if esDataType==None:
          raise ValueError(f"Column:{itemName}, MySQL Type:{dataType} -> esDataType is null!")
        dataInfoMap[itemName]={"type":esDataType}
  resultData={"mappings":{"properties":dataInfoMap}}
  print(f"PUT /{tableName}\n{json.dumps(resultData)}")


if __name__ == '__main__':
  trans()