python to elasticsearch

파이선에서 elk로 연결하고 bulk update 하는 방법을 알아보자.

기본적으로 ES가 잘 접속 되는지 확인한다.

!pip install elasticsearch

from elasticsearch import Elasticsearch
from elasticsearch import helpers

es_client = Elasticsearch("https://localhost:9200/", verify_certs=False)
es_client.info()

데이터 null 처리와 날짜 형식 예외 처리 함수가 필요하다.

from elasticsearch import Elasticsearch
from elasticsearch import helpers
from datetime import datetime

def safe_date(date_value):
    return (
        pd.to_datetime(date_value) if not pd.isna(date_value)
            else  datetime(1970,1,1,0,0)
    )
df['ImportantDate'] = df['ImportantDate'].apply(safe_date)

def safe_value(field_val):
    return field_val if not pd.isna(field_val) else "Other"
df['Hold'] = df['PossiblyBlankField'].apply(safe_value)

df_iter = df.iterrows()
index, document = next(df_iter)

use_these_keys = ['id', 'FirstName', 'LastName', 'ImportantDate']
def filterKeys(document):
    return {key: document[key] for key in use_these_keys }
    
    es_client = Elasticsearch(http_compress=True)
def doc_generator(df):
    df_iter = df.iterrows()
    for index, document in df_iter:
        yield {
                "_index": 'your_index',
                "_type": "_doc",
                "_id" : f"{document['id']}",
                "_source": filterKeys(document),
            }
    raise StopIteration
helpers.bulk(es_client, doc_generator(your_dataframe))

 

<첨부 링크>

https://towardsdatascience.com/exporting-pandas-data-to-elasticsearch-724aa4dd8f62

https://doubly12f.tistory.com/46

https://blog.naver.com/PostView.nhn?isHttpsRedirect=true&blogId=wideeyed&logNo=221494109911