csv2elastic.py

import os
import csv
import types
import asyncio
import json
import aiohttp



async def feed_elastic():
    exhausted = False

    @asyncio.coroutine
    def produce_elastic_commands(reader, num):
        nonlocal exhausted
        for index, row in enumerate(reader):
            print(index)
            yield b'{"create":{}}\n'
            yield json.dumps(row).encode('utf-8')
            yield b'\n'
            if index > num:
                return
        exhausted = True

    async with aiohttp.ClientSession() as session:
        with open('llll_data.csv') as inp:
            size = os.stat(inp.fileno()).st_size
            reader = csv.DictReader(inp)
            while True:
                streamer = produce_elastic_commands(reader, 100000)
                response = await session.post('http://localhost:9200/llll/sheet/_bulk', data=streamer)
                if exhausted:
                    break
                print(response)
                await reponse


asyncio.get_event_loop().run_until_complete(feed_elastic())