<commit_before>###############################################################################
#
# Flume / Kafka ingest pipeline, offering 3 broadcast channels that can be 
# tapped with differing qualities of service:
#
#   1. single/non-durable, optimised for lowest latency delivery of single events,
#      add memory channel to mydataset.sources.source_single.channels
#   2. single/durable, optimised for low latency delivery of single events with
#      durable operation, add kafka source to topic mydataset_single
#   3. batch/durable, optimised for greatest throughput of batch events with
#      durable operation, add kafka source to topic mydataset_batch
#
###############################################################################

mydataset.sources=source_single source_batch
mydataset.channels=channel_single_nondurable_hdfs channel_single_durable channel_batch_durable
mydataset.sinks=sink_single_hdfs sink_batch_hdfs

mydataset.sources.source_single.type=com.cloudera.example.stream.Stream
mydataset.sources.source_single.recordType=xml
mydataset.sources.source_single.pollMs=10000
mydataset.sources.source_single.pollTicks=0
mydataset.sources.source_single.batchSize=1
mydataset.sources.source_single.recordNumber=10
mydataset.sources.source_single.interceptors=interceptor_stream
mydataset.sources.source_single.interceptors.interceptor_stream.type=com.cloudera.example.stream.Stream$Interceptor$Builder
mydataset.sources.source_single.selector.type=replicating
mydataset.sources.source_single.channels=channel_single_nondurable_hdfs channel_single_durable

mydataset.channels.channel_single_nondurable_hdfs.type=memory
mydataset.channels.channel_single_nondurable_hdfs.capacity=10000
mydataset.channels.channel_single_nondurable_hdfs.transactionCapacity=1000
mydataset.channels.channel_single_nondurable_hdfs.keep-alive=1
mydataset.channels.channel_single_nondurable_hdfs.byteCapacityBufferPercentage=1

mydataset.sinks.sink_single_hdfs.type=hdfs
mydataset.sinks.sink_single_hdfs.hdfs.path=$ROOT_HDFS/$ROOT_DIR_HDFS_RAW/source/text/%{bt}/none/ingest_batch=%{btss}_%{btsf}_mydataset-%{bid}.%{bt}
mydataset.sinks.sink_single_hdfs.hdfs.filePrefix=%{ts}_mydataset-%{bi}-of-%{bs}
mydataset.sinks.sink_single_hdfs.hdfs.inUsePrefix=_
mydataset.sinks.sink_single_hdfs.hdfs.rollCount=0
mydataset.sinks.sink_single_hdfs.hdfs.rollInterval=0
mydataset.sinks.sink_single_hdfs.hdfs.rollSize=0
mydataset.sinks.sink_single_hdfs.hdfs.idleTimeout=1
mydataset.sinks.sink_single_hdfs.hdfs.batchSize=1
mydataset.sinks.sink_single_hdfs.hdfs.fileType=DataStream
mydataset.sinks.sink_single_hdfs.channel=channel_single_nondurable_hdfs

mydataset.channels.channel_single_durable.type=org.apache.flume.channel.kafka.KafkaChannel
mydataset.channels.channel_single_durable.brokerList=$KAFKA_KAFKA_BROKER_HOSTS_AND_PORTS
mydataset.channels.channel_single_durable.zookeeperConnect=$ZOOKEEPER_SERVER_HOSTS_AND_PORTS
mydataset.channels.channel_single_durable.capacity=10000
mydataset.channels.channel_single_durable.transactionCapacity=1000
mydataset.channels.channel_single_durable.topic=mydataset_single

mydataset.sources.source_batch.type=org.apache.flume.source.kafka.KafkaSource
mydataset.sources.source_batch.zookeeperConnect=$ZOOKEEPER_SERVER_HOSTS_AND_PORTS
mydataset.sources.source_batch.groupId=source_batch
mydataset.sources.source_batch.topic=mydataset_single
mydataset.sources.source_batch.batchSize=10
mydataset.sources.source_batch.batchDurationMillis=700000
mydataset.sources.source_batch.interceptors=interceptor_unwrap interceptor_stream
mydataset.sources.source_batch.interceptors.interceptor_unwrap.type=com.cloudera.framework.main.common.flume.FlumeEventUnwrapInterceptor$Builder
mydataset.sources.source_batch.interceptors.interceptor_stream.type=com.cloudera.example.stream.Stream$Interceptor$Builder
mydataset.sources.source_batch.channels=channel_batch_durable

mydataset.channels.channel_batch_durable.type=org.apache.flume.channel.kafka.KafkaChannel
mydataset.channels.channel_batch_durable.brokerList=$KAFKA_KAFKA_BROKER_HOSTS_AND_PORTS
mydataset.channels.channel_batch_durable.zookeeperConnect=$ZOOKEEPER_SERVER_HOSTS_AND_PORTS
mydataset.channels.channel_batch_durable.capacity=10000
mydataset.channels.channel_batch_durable.transactionCapacity=1000
mydataset.channels.channel_batch_durable.topic=mydataset_batch

mydataset.sinks.sink_batch_hdfs.type=hdfs
mydataset.sinks.sink_batch_hdfs.hdfs.path=$ROOT_HDFS/$ROOT_DIR_HDFS_STAGED/partitioned/sequence/%{bt}/none/ingest_batch_id=%{bid}/ingest_batch_start=%{btss}/ingest_batch_finish=%{btsf}
mydataset.sinks.sink_batch_hdfs.hdfs.filePrefix=mydataset
mydataset.sinks.sink_batch_hdfs.hdfs.inUsePrefix=_
mydataset.sinks.sink_batch_hdfs.hdfs.rollCount=0
mydataset.sinks.sink_batch_hdfs.hdfs.rollInterval=0
mydataset.sinks.sink_batch_hdfs.hdfs.rollSize=0
mydataset.sinks.sink_batch_hdfs.hdfs.idleTimeout=110
mydataset.sinks.sink_batch_hdfs.hdfs.batchSize=10
mydataset.sinks.sink_batch_hdfs.hdfs.writeFormat=com.cloudera.example.stream.Stream$Serializer$Builder
mydataset.sinks.sink_batch_hdfs.hdfs.fileType=SequenceFile
mydataset.sinks.sink_batch_hdfs.channel=channel_batch_durable<commit_msg>Set batch write idletimeout to short delay<commit_after>###############################################################################
#
# Flume / Kafka ingest pipeline, offering 3 broadcast channels that can be 
# tapped with differing qualities of service:
#
#   1. single/non-durable, optimised for lowest latency delivery of single events,
#      add memory channel to mydataset.sources.source_single.channels
#   2. single/durable, optimised for low latency delivery of single events with
#      durable operation, add kafka source to topic mydataset_single
#   3. batch/durable, optimised for greatest throughput of batch events with
#      durable operation, add kafka source to topic mydataset_batch
#
###############################################################################

mydataset.sources=source_single source_batch
mydataset.channels=channel_single_nondurable_hdfs channel_single_durable channel_batch_durable
mydataset.sinks=sink_single_hdfs sink_batch_hdfs

mydataset.sources.source_single.type=com.cloudera.example.stream.Stream
mydataset.sources.source_single.recordType=xml
mydataset.sources.source_single.pollMs=10000
mydataset.sources.source_single.pollTicks=0
mydataset.sources.source_single.batchSize=1
mydataset.sources.source_single.recordNumber=10
mydataset.sources.source_single.interceptors=interceptor_stream
mydataset.sources.source_single.interceptors.interceptor_stream.type=com.cloudera.example.stream.Stream$Interceptor$Builder
mydataset.sources.source_single.selector.type=replicating
mydataset.sources.source_single.channels=channel_single_nondurable_hdfs channel_single_durable

mydataset.channels.channel_single_nondurable_hdfs.type=memory
mydataset.channels.channel_single_nondurable_hdfs.capacity=10000
mydataset.channels.channel_single_nondurable_hdfs.transactionCapacity=1000
mydataset.channels.channel_single_nondurable_hdfs.keep-alive=1
mydataset.channels.channel_single_nondurable_hdfs.byteCapacityBufferPercentage=1

mydataset.sinks.sink_single_hdfs.type=hdfs
mydataset.sinks.sink_single_hdfs.hdfs.path=$ROOT_HDFS/$ROOT_DIR_HDFS_RAW/source/text/%{bt}/none/ingest_batch=%{btss}_%{btsf}_mydataset-%{bid}.%{bt}
mydataset.sinks.sink_single_hdfs.hdfs.filePrefix=%{ts}_mydataset-%{bi}-of-%{bs}
mydataset.sinks.sink_single_hdfs.hdfs.inUsePrefix=_
mydataset.sinks.sink_single_hdfs.hdfs.rollCount=0
mydataset.sinks.sink_single_hdfs.hdfs.rollInterval=0
mydataset.sinks.sink_single_hdfs.hdfs.rollSize=0
mydataset.sinks.sink_single_hdfs.hdfs.idleTimeout=1
mydataset.sinks.sink_single_hdfs.hdfs.batchSize=1
mydataset.sinks.sink_single_hdfs.hdfs.fileType=DataStream
mydataset.sinks.sink_single_hdfs.channel=channel_single_nondurable_hdfs

mydataset.channels.channel_single_durable.type=org.apache.flume.channel.kafka.KafkaChannel
mydataset.channels.channel_single_durable.brokerList=$KAFKA_KAFKA_BROKER_HOSTS_AND_PORTS
mydataset.channels.channel_single_durable.zookeeperConnect=$ZOOKEEPER_SERVER_HOSTS_AND_PORTS
mydataset.channels.channel_single_durable.capacity=10000
mydataset.channels.channel_single_durable.transactionCapacity=1000
mydataset.channels.channel_single_durable.topic=mydataset_single

mydataset.sources.source_batch.type=org.apache.flume.source.kafka.KafkaSource
mydataset.sources.source_batch.zookeeperConnect=$ZOOKEEPER_SERVER_HOSTS_AND_PORTS
mydataset.sources.source_batch.groupId=source_batch
mydataset.sources.source_batch.topic=mydataset_single
mydataset.sources.source_batch.batchSize=10
mydataset.sources.source_batch.batchDurationMillis=700000
mydataset.sources.source_batch.interceptors=interceptor_unwrap interceptor_stream
mydataset.sources.source_batch.interceptors.interceptor_unwrap.type=com.cloudera.framework.main.common.flume.FlumeEventUnwrapInterceptor$Builder
mydataset.sources.source_batch.interceptors.interceptor_stream.type=com.cloudera.example.stream.Stream$Interceptor$Builder
mydataset.sources.source_batch.channels=channel_batch_durable

mydataset.channels.channel_batch_durable.type=org.apache.flume.channel.kafka.KafkaChannel
mydataset.channels.channel_batch_durable.brokerList=$KAFKA_KAFKA_BROKER_HOSTS_AND_PORTS
mydataset.channels.channel_batch_durable.zookeeperConnect=$ZOOKEEPER_SERVER_HOSTS_AND_PORTS
mydataset.channels.channel_batch_durable.capacity=10000
mydataset.channels.channel_batch_durable.transactionCapacity=1000
mydataset.channels.channel_batch_durable.topic=mydataset_batch

mydataset.sinks.sink_batch_hdfs.type=hdfs
mydataset.sinks.sink_batch_hdfs.hdfs.path=$ROOT_HDFS/$ROOT_DIR_HDFS_STAGED/partitioned/sequence/%{bt}/none/ingest_batch_id=%{bid}/ingest_batch_start=%{btss}/ingest_batch_finish=%{btsf}
mydataset.sinks.sink_batch_hdfs.hdfs.filePrefix=mydataset
mydataset.sinks.sink_batch_hdfs.hdfs.inUsePrefix=_
mydataset.sinks.sink_batch_hdfs.hdfs.rollCount=0
mydataset.sinks.sink_batch_hdfs.hdfs.rollInterval=0
mydataset.sinks.sink_batch_hdfs.hdfs.rollSize=0
mydataset.sinks.sink_batch_hdfs.hdfs.idleTimeout=1
mydataset.sinks.sink_batch_hdfs.hdfs.batchSize=10
mydataset.sinks.sink_batch_hdfs.hdfs.writeFormat=com.cloudera.example.stream.Stream$Serializer$Builder
mydataset.sinks.sink_batch_hdfs.hdfs.fileType=SequenceFile
mydataset.sinks.sink_batch_hdfs.channel=channel_batch_durable
{'fraction_non_alphanumeric': 0.11131510290849357, 'fraction_numerical': 0.011981833993622572, 'mean_word_length': 10.213434452871073, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '7256317', 'n_tokens_mistral': 3558, 'n_tokens_neox': 3449, 'n_words': 283}