چطور میتونم دیتا رو از دیتابیس به صورت bulk وارد elasticsearch کنم؟ اگربا elasticsearch_dsl بشود خیلی بهتر است

Question

در صورتی که اطلاعات موجود در elasticsearch از بین برود چطور می تونم اطلاعات رو به صورت bulk از دیتابیسم که مثلا postgres هست وارد elasticsearch کنم؟

 برای اینکار در مدل های postgres یک فانکشن نوشتم  to_es که دیتا رو به صورت json در بیارم

می دونم که باید از bulk api elasticsearch  استفاده کنم اما میخوام به صورت فانکشن داخل پروژه بنویسم

لطفا راهنماییم کنید.

Answer 1

اگر از جاوا استفاده می کنی که elasticsearch-client که جزو ماژولای native الستیک  هست.  به پروژت اضافه می کنی و استفاده می کنی.

اگر از spring  استفاده می کنید در پروژتون spring-data-elasticsearch گزینه خوبی هست.

ولی گزینه های دیگه هم دارید مثل jest

https://github.com/searchbox-io/Jest

فرق jest با اون دوتای دیگه اینه که jest سرویسای Rest صدا می زنه و خوب روی http ولی اون دو تا مثل ارتباط دو تا node هست و روی http نیست.

اگر از زبان دیگه ای استفاده می کنیدخوب Elasticsearch یک سری سرویسهای Rest  (http://localhost:9200/anIndex/aType/_bulk) در اختیار شما قرار می ده که می تونی از داخل برنامت اونارو صدا بزنی  و ایندکس کنی البته می تونی از پروژه های زیر (غیر جاوایی)نیز استفاده کنید.

https://elasticsearch-py.readthedocs.org/en/master   

https://github.com/elastic/elasticsearch-php

https://github.com/elastic/elasticsearch-ruby

Comments

ممنون از پاسختون :) . از helper خود elasticsearch استفاده کردم.

Answer 2

چون میخواستم به صورت یک api و فانکشن باشه از helper استفاده کردم .

from elasticsearch import helpers, Elasticsearch

es = Elasticsearch()

in my model: (Sqlalchemy)

    def to_es_details(self):
        details = self.data
        data = {}

        data.update({"_index": "library:book_details", "_type": "book_details", "_id": self.id,
                     "_source": {'title': self.title, 'publisher': self.publisher_uid, 'lang': self.language,
                                 'category': self.category,  'download_count': self.download_count, 'price': self.price, 'discount': self.discount, 'created_at': self.created_epoch, 'status': self.status, 'view_count': self.view_count, 'rate': self.rate}})
        return data


​
def bulk_pg_to_es():
    media = session.query(Media).filter(Media.status != u'EXPIRED').all()
    data = []
    for m in media:
        data.append(m.to_es_details())
    helpers.bulk(es, data)