(pillow_id='ApplicationToElasticsearchPillow', num_processes=1,
                                    process_num=0, **kwargs)
return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=app_processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
    )
App pillow

    Processors:
      - :py:class:`pillowtop.processors.elastic.BulkElasticProcessor`
App pillow
App pillow
def get_app_to_elasticsearch_pillow(pillow_id='ApplicationToElasticsearchPillow', num_processes=1,
                                    process_num=0, **kwargs):
    """App pillow

    Processors:
      - :py:class:`pillowtop.processors.elastic.BulkElasticProcessor`
    """
    assert pillow_id == 'ApplicationToElasticsearchPillow', 'Pillow ID is not allowed to change'
    checkpoint = get_checkpoint_for_elasticsearch_pillow(pillow_id, APP_INDEX_INFO, [topics.APP])
    app_processor = ElasticProcessor(
        elasticsearch=get_es_new(),
        index_info=APP_INDEX_INFO,
        doc_prep_fn=transform_app_for_es
    )
    change_feed = KafkaChangeFeed(
        topics=[topics.APP], client_id='apps-to-es', num_processes=num_processes, process_num=process_num
    )
    return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=app_processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
    )
def get_app_to_elasticsearch_pillow ( pillow_id = 'ApplicationToElasticsearchPillow' , num_processes = 1 , process_num = 0 , * * kwargs ) : assert pillow_id == 'ApplicationToElasticsearchPillow' , 'Pillow ID is not allowed to change' checkpoint = get_checkpoint_for_elasticsearch_pillow ( pillow_id , APP_INDEX_INFO , [ topics . APP ] ) app_processor = ElasticProcessor ( elasticsearch = get_es_new ( ) , index_info = APP_INDEX_INFO , doc_prep_fn = transform_app_for_es ) change_feed = KafkaChangeFeed ( topics = [ topics . APP ] , client_id = 'apps-to-es' , num_processes = num_processes , process_num = process_num ) return ConstructedPillow ( name = pillow_id , checkpoint = checkpoint , change_feed = change_feed , processor = app_processor , change_processed_event_handler = KafkaCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 100 , change_feed = change_feed ) , )