Load Tweets to Elasticsearch using Pandas and Python Elasticsearch client

02_load_tweets_es.py

import tqdm
from elasticsearch import Elasticsearch
from elasticsearch.helpers import streaming_bulk
import pandas as pd

FILE_LOC = 'staging/TweetsElonMusk.csv'
INDEX_NAME = 'elonmusktweets'
df = pd.read_csv(FILE_LOC)


tweets_df = (df
.dropna(axis=1, how='any')
.drop(columns=['timezone', 'user_id', 'username', 'name', 'mentions', 'urls', 'photos', 'hashtags', 'cashtags', 'video', 'created_at'])
)
tweets_df['date'] = pd.to_datetime(tweets_df['date'])
tweets_df['_id'] = tweets_df['id']
tweets_df.drop(columns=['id'])

print(tweets_df.info())

tweets = tweets_df.to_dict(orient='records')
print(len(tweets))


def get_tweet_details():
    for tweet in tweets:
        yield tweet


number_of_tweets = len(tweets)
print(f"total number_of_tweets: {number_of_tweets}")


es_client = Elasticsearch(hosts='localhost')

print("Indexing tweets...")
progress = tqdm.tqdm(unit="tweets", total=number_of_tweets)
successes = 0
for ok, action in streaming_bulk(client=es_client, index=INDEX_NAME, actions=get_tweet_details()):
    progress.update(1)
    successes += ok

print("Indexed %d/%d tweets successfully" % (successes, number_of_tweets))