(pillow_id='SqlSMSPillow', num_processes=1, process_num=0,
                       processor_chunk_size=DEFAULT_PROCESSOR_CHUNK_SIZE, **kwargs)
return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
        processor_chunk_size=processor_chunk_size
    )
SMS Pillow

    Processors:
      - :py:class:`pillowtop.processors.elastic.BulkElasticProcessor`
SMS Pillow
SMS Pillow
def get_sql_sms_pillow(pillow_id='SqlSMSPillow', num_processes=1, process_num=0,
                       processor_chunk_size=DEFAULT_PROCESSOR_CHUNK_SIZE, **kwargs):
    """SMS Pillow

    Processors:
      - :py:class:`pillowtop.processors.elastic.BulkElasticProcessor`
    """
    assert pillow_id == 'SqlSMSPillow', 'Pillow ID is not allowed to change'
    checkpoint = get_checkpoint_for_elasticsearch_pillow(pillow_id, SMS_INDEX_INFO, [topics.SMS])
    processor = BulkElasticProcessor(
        elasticsearch=get_es_new(),
        index_info=SMS_INDEX_INFO,
        doc_prep_fn=lambda x: x
    )
    change_feed = KafkaChangeFeed(
        topics=[topics.SMS], client_id='sql-sms-to-es',
        num_processes=num_processes, process_num=process_num
    )
    return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
        processor_chunk_size=processor_chunk_size
    )
def get_sql_sms_pillow ( pillow_id = 'SqlSMSPillow' , num_processes = 1 , process_num = 0 , processor_chunk_size = DEFAULT_PROCESSOR_CHUNK_SIZE , * * kwargs ) : assert pillow_id == 'SqlSMSPillow' , 'Pillow ID is not allowed to change' checkpoint = get_checkpoint_for_elasticsearch_pillow ( pillow_id , SMS_INDEX_INFO , [ topics . SMS ] ) processor = BulkElasticProcessor ( elasticsearch = get_es_new ( ) , index_info = SMS_INDEX_INFO , doc_prep_fn = lambda x : x ) change_feed = KafkaChangeFeed ( topics = [ topics . SMS ] , client_id = 'sql-sms-to-es' , num_processes = num_processes , process_num = process_num ) return ConstructedPillow ( name = pillow_id , checkpoint = checkpoint , change_feed = change_feed , processor = processor , change_processed_event_handler = KafkaCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 100 , change_feed = change_feed ) , processor_chunk_size = processor_chunk_size )