实现es的mysql中的upsert

最近由于查询速度要求。原使用的python转而使用es进行全文检索查询。

关于标签存储，es存储是一种选择。

原使用某国产数据库推送来的数据，数据字段需要与采集系统数据字段保持一致。所以有些字段例如RANK属于关键字。

脚本语言使用的是python3，所以在使用pymysql时会报错。因为将字段名称识别为rank函数。所以需要        表名.rank        的形式。

 通过建立mysql连接,获取mysql表内数据，数据形式为tuple

def readmysql(sql):
   
    conn = pymysql.connect(host='localhost',
                         user='username',
                         password='your password',
                         database='databasename',
                         charset='utf8',
                         port=portnum)
 
    result = pd.read_sql(sql, conn)
    result=list(result.itertuples(index=False))
    return result

第二步，写入es。使用方法为upsert

def writetoes(val):
    es = Elasticsearch([{'host':'hostaddress','port':portnum}],
                       sniff_on_start=False,
                       sniff_on_connection_fail=True,
                       sniffer_timeout=60000,
                       sniff_timeout=5000,
                       request_timeout=500000)
   
    
    for j in range(0,len(val),1000):
        actions = []
        bound=min(j+1000,len(val))
        for i in range(j,bound):
            action = {
                    "_index": "indexname",
                    "_type": "_doc",
                    "_id":val[i][0],
                    '_op_type': 'update',
                    "doc_as_upsert":True,
                    "null_value":" ",
                    "doc": {
                        
                        "upsert1": val[i][0],
                        "upsert2": val[i][1],
                        "upsert3":val[i][2],

                    }
            }
            actions.append(action)
        helpers.bulk(es, actions,request_timeout=500000)

 其中具体的参数需要自行进行配置，写入es采用的是bulk以及helper。一千条一次进行插入。request_timeout需要自行根据连接时间进行调整。

起到upsert关键作用的参数为doc_as_upsert，将其设为TRUE

es读取全量数据方法scroll方法相较于bulk会慢上许多。可能还需要进行调整优化。优化完再写，优化不出来就不写。