<commit_before># -*- coding: utf-8 -*-

from elasticsearch import Elasticsearch
from elasticsearch.client import IndicesClient
from elasticsearch.exceptions import NotFoundError
from elasticsearch.exceptions import TransportError
from storm import Bolt, log


class ESIndexBolt(Bolt):
    def initialize(self, conf, context):
        log('Bolt id-%s is initializing.' % id(self))
        # TODO: Make connection params configurable.
        self.es = Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
        ic = IndicesClient(self.es)

        try:
            ic.get_mapping(
                index='observations',
                doc_type='user'
                )
        except NotFoundError:
            doc_type = {
                'user': {
                    'properties': {
                        'events': {
                            'type': 'nested'
                            }
                        }
                    }
                }
            ic.put_mapping(
                index='observations',
                ignore_conflicts=True,
                doc_type='user',
                body=doc_type
                )

    def process(self, tup):
        event = dict(
            timestamp=tup.values[0],
            path=tup.values[1]
            )

        params = dict(
            index='observations',
            id=tup.values[2],
            doc_type='user'
            )

        try:
            events = self.es.get(**params)['_source']['events']
            params['body'] = {'events': events + [event]}
        except NotFoundError:
            params['body'] = {'events': [event]}
            params['op_type'] = 'create'
        except TransportError, e:
            # TODO: What is going wrong here?
            log('[TransportError] %s' % e)

        self.es.index(**params)

ESIndexBolt().run()<commit_msg>Don't crash on Transport Errors<commit_after># -*- coding: utf-8 -*-

from elasticsearch import Elasticsearch
from elasticsearch.client import IndicesClient
from elasticsearch.exceptions import NotFoundError
from elasticsearch.exceptions import TransportError
from storm import Bolt, log


class ESIndexBolt(Bolt):
    def initialize(self, conf, context):
        log('Bolt id-%s is initializing.' % id(self))
        # TODO: Make connection params configurable.
        self.es = Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
        ic = IndicesClient(self.es)

        try:
            ic.get_mapping(
                index='observations',
                doc_type='user'
                )
        except NotFoundError:
            doc_type = {
                'user': {
                    'properties': {
                        'events': {
                            'type': 'nested'
                            }
                        }
                    }
                }
            ic.put_mapping(
                index='observations',
                ignore_conflicts=True,
                doc_type='user',
                body=doc_type
                )

    def process(self, tup):
        event = dict(
            timestamp=tup.values[0],
            path=tup.values[1]
            )

        kwargs = dict(
            id=tup.values[2]
            )

        try:
            events = self.es.get('observations', tup.values[2], 'user')
            body = {'events': events['_source']['events'] + [event]}
        except NotFoundError:
            kwargs['op_type'] = 'create'
            body = {'events': [event]}
        except TransportError, e:
            # TODO: What is going wrong here?
            log('[TransportError] Get failed: %s' % e)
            return

        try:
            self.es.index('observations', 'user', body, **kwargs)
        except TransportError, e:
            # TODO: What is going wrong here?
            log('[TransportError] Index failed: %s' % e)

ESIndexBolt().run()
{'fraction_non_alphanumeric': 0.11241997439180537, 'fraction_numerical': 0.004353393085787452, 'mean_word_length': 1.4612476370510397, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2991461', 'n_tokens_mistral': 1051, 'n_tokens_neox': 1001, 'n_words': 232}