(pillow_id='KafkaDomainPillow', num_processes=1,
                                             process_num=0, **kwargs)
return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=domain_processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
    )
Domain pillow to replicate documents to ES

    Processors:
      - :py:class:`pillowtop.processors.elastic.ElasticProcessor`
Domain pillow to replicate documents to ES
Domain pillow to replicate documents to ES
def get_domain_kafka_to_elasticsearch_pillow(pillow_id='KafkaDomainPillow', num_processes=1,
                                             process_num=0, **kwargs):
    """Domain pillow to replicate documents to ES

    Processors:
      - :py:class:`pillowtop.processors.elastic.ElasticProcessor`
    """
    assert pillow_id == 'KafkaDomainPillow', 'Pillow ID is not allowed to change'
    checkpoint = get_checkpoint_for_elasticsearch_pillow(pillow_id, DOMAIN_INDEX_INFO, [topics.DOMAIN])
    domain_processor = ElasticProcessor(
        elasticsearch=get_es_new(),
        index_info=DOMAIN_INDEX_INFO,
        doc_prep_fn=transform_domain_for_elasticsearch,
    )
    change_feed = KafkaChangeFeed(
        topics=[topics.DOMAIN], client_id='domains-to-es', num_processes=num_processes, process_num=process_num
    )
    return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=domain_processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
    )
def get_domain_kafka_to_elasticsearch_pillow ( pillow_id = 'KafkaDomainPillow' , num_processes = 1 , process_num = 0 , * * kwargs ) : assert pillow_id == 'KafkaDomainPillow' , 'Pillow ID is not allowed to change' checkpoint = get_checkpoint_for_elasticsearch_pillow ( pillow_id , DOMAIN_INDEX_INFO , [ topics . DOMAIN ] ) domain_processor = ElasticProcessor ( elasticsearch = get_es_new ( ) , index_info = DOMAIN_INDEX_INFO , doc_prep_fn = transform_domain_for_elasticsearch , ) change_feed = KafkaChangeFeed ( topics = [ topics . DOMAIN ] , client_id = 'domains-to-es' , num_processes = num_processes , process_num = process_num ) return ConstructedPillow ( name = pillow_id , checkpoint = checkpoint , change_feed = change_feed , processor = domain_processor , change_processed_event_handler = KafkaCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 100 , change_feed = change_feed ) , )