org.thales.punch.libraries.storm.bolt

Class AbstractFileOutput

java.lang.Object

org.thales.punch.libraries.storm.api.BaseProcessingNode

org.thales.punch.libraries.storm.bolt.AbstractFileOutput

All Implemented Interfaces:

Serializable, org.apache.storm.task.IBolt, org.apache.storm.topology.IComponent, org.apache.storm.topology.IRichBolt, org.thales.punch.libraries.punchlang.api.IRuntimeContext

Direct Known Subclasses:

AtLeastOnceFileOutput, OneByOneFileOutput

public abstract class AbstractFileOutput
extends org.thales.punch.libraries.storm.api.BaseProcessingNode
implements org.thales.punch.libraries.punchlang.api.IRuntimeContext

The file bolt writes the incoming tuples into archives. It batches the tuples in groups, flushes the whole content at once in the configured archive cluster.

The file bolt settings lets you decide where to flush the batches (filesystem or ceph objects) and act on the saved archive format (compression, encryption, separator etc..).

To take into consideration that cluster parameter can take multiple mount points, each of them should be delimited be a comma separator. For instance:
"cluster": "file:///tmp/archive-logs-1/storage,file:///tmp/archive-logs-2/storage"

Here is a complete example to flush the batch to a (local or mounted) filesystem:

    "bolts" : [
         {
             "type" : "file_bolt",
             "bolt_settings" : {
               "cluster" : "file:///tmp/archive",
             }
             "storm_settings" : {
                 "executors": 1,
                  "subscribe" : [ { "stream" : "logs" } ]
             }
         }
  

Settings

property	mandatory	type	default	comment
"cluster"	yes	String	-	The url of the destination archive store. See below for details
"compression_format"	no	String	"NONE"	One of "NONE", "SNAPPY" or "GZIP"
"folders_hierarchy_pattern"	no	String	"%{topic}/%{partition}/%{date}/%{batches_set}"	If objects-storage is implemented over a file-system, batches of logs will be written on multiple folders and sub-folders. This folders hierarchy is described in this parameter as a pattern. It can contain a static string like "my_archives/my_technology" and/or specific patterns: %{topic} for a topic, %{partition} for partition number, %{date} for earliest tuple date time in batch (following YYYY.mm.dd format only), or %{batches_set} which is the first numbers of batch number. This last pattern is tricky: it allows to store a fixed maximum number of batches in a folder. For example if "batches_per_folder" equals 1000 and if "folders_hierarchy_pattern" is "%{batches_set} then batches 1098 and 1099 will be stored in folder "1" and batches 2000 and 2001 will be stored in folder "2".
"batches_per_folder"	no	Integer	1000	If objects-storage is implemented over a file-system, batches of logs will be written on multiple folders and sub-folders. Each sub-folder, if default folders hierarchy pattern is kept, will contain at most this number of batches.
"strategy"	no	String	"exactly_once"	Define the batching strategy, accepted values are: "exactly_once", "at_least_once", "one_by_one"
"create_root"	no	boolean	false	By default, the filebolt does not create the root directory. Set this option to true will create the root directory if it doesn't exists.
"tags"	no	String[]	-	Only works in "at_least_once" batching mode. This parameter modify the batch grouping strategy to regroup events based on their "tags" values. A subscribed Storm field must correspond to each tag. The bolt reads the values contained in each field and regroups all the Tuples with the same values in the same batch. All the tags values are appended to the final file name.
"batch_size"	no	Integer	1000	Only for "at_least_once" case, define the maximum number of events stored in a batch.
"batch_expiration_timeout"	no	String	10s	Only for "at_least_once" case. If no new events are added to a batch for more than this period, the batch is flushed.
"batch_life_timeout"	no	String	0	Only for "at_least_once" case. If the batch has been alive for more than this period, the batch is flushed.

Storage Backends

The cluster parameter lets you choose among the following archive endpoints :

• "file:///path": use the object storage archive file format (like for ceph) and save the object as regular file to the (shared) filesystem. The path must point to an absolute folder name.
• "ceph_configuration://path" : use the ceph archive. The path must point to an explicit ceph configuration file.

Archive Indexing

This bolt emits metadata containing key information about each batch (compressed size, encoding...). By cascading an ElasticBolt, you can index these metadata in ElasticSearch.

Archive Format

The archive files or objects (if using ceph object storage) contains basically csv data, containing all or the configured the received fields.

Archive Retention

If you use the retention property, the archives will be generated using a cyclic ring naming scheme in a way to ultimately overwrites expired archives with the new ones. This releases you to remove expired archive which can turn out to be a costly operation.

Batching Strategies

The Filebolt has several batching strategies: "exactly_once", "at_least_once" or "one_by_one".
- "exactly_once": ensure that any event is only written once. If a topology failure occurs the latest bucket will be replayed and overwritten. Input must be from Kafka spout. The batch size is configured at the entry spout level, refer to the BatchKafkaSpout documentation. Flushes upon receiving an end-of-batch signal from the kafka.
- "at_least_once": ensure that any event is stored at least one time (maybe more). If the topology fails, new files will be created (no overwrite) which is especially useful for filesystem that lock files. Input must be from Kafka spout. The batch size is configured at the filebolt level. Flushes when reaching batch size or expiration time.
- "one_by_one": the batch size is set to one, each event will be written in a dedicated file.

Check for additional documentation: https://doc.punchplatform.com/Reference_Guide/Data_Movement/Bolts/File_Bolt.html

Author:

D. Tombroff/C.Van Frachem

See Also:

AtLeastOnceFileBolt, OneByOneFileBolt, Serialized Form

Field Summary

Fields

Modifier and Type	Field and Description
protected org.thales.punch.libraries.objectstorage.tuples.ArchiveFormatSettings	archiveFormatSettings The archive format settings.
protected IFileNameSupplier	fileNameSupplier This supplier is in charge of providing us with the identifier of each batch.
protected com.codahale.metrics.Timer	tupleRtt Time to process tuple metric
protected org.thales.punch.libraries.storm.api.StreamDeclaration	userStream The user stream where to publish the batches metadata.

Fields inherited from class org.thales.punch.libraries.storm.api.BaseProcessingNode

ackRate, collector, componentId, every, failRate, metricContext, nodeSettings, topologyContext, traversalTime

Constructor Summary

Constructors

Constructor and Description
AbstractFileOutput(org.thales.punch.libraries.storm.api.NodeSettings boltConfig, List<org.thales.punch.libraries.objectstorage.tuples.ArchiveDevice> archiveDevices, IFileNameSupplier fileNameSupplier, org.thales.punch.libraries.objectstorage.tuples.ArchiveFormatSettings archiveFormatSettings) Ctor.

Method Summary

Modifier and Type	Method and Description
void	addRuntimeData(Object context, int key, Object value)
protected BatchArchiver	createArchiver(org.thales.punch.libraries.objectstorage.indexing.BatchMetadata metadata) Create correct BatchArchiver from metadata
protected void	flush(BatchArchiver batch) Flush this batch, i.e.
List<String>	getBloomFilterFields() Getter for bloom fields.
Object	getRuntimeData(Object context, int key)
long	getTimestampFromTuple(org.apache.storm.tuple.ITuple tuple) Extracts a timestamp from the received tuple.
protected org.thales.punch.libraries.objectstorage.indexing.BatchMetadata	initMetadata(long timestamp) Common metadata initialization
void	prepare(Map stormConf, org.apache.storm.task.TopologyContext context, org.apache.storm.task.OutputCollector outputCollector) Prepare this bolt.

Methods inherited from class org.thales.punch.libraries.storm.api.BaseProcessingNode

ack, cleanup, declareOutputFields, enrichAndForwardMonitoringTuple, execute, fail, getComponentConfiguration, getNewPoint, process

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

userStream

protected org.thales.punch.libraries.storm.api.StreamDeclaration userStream

The user stream where to publish the batches metadata.

archiveFormatSettings

protected final org.thales.punch.libraries.objectstorage.tuples.ArchiveFormatSettings archiveFormatSettings

The archive format settings. I.e. csv or not, delimiter, fields etc..

fileNameSupplier

protected final IFileNameSupplier fileNameSupplier

This supplier is in charge of providing us with the identifier of each batch.

tupleRtt

protected transient com.codahale.metrics.Timer tupleRtt

Time to process tuple metric

Constructor Detail

AbstractFileOutput

public AbstractFileOutput(org.thales.punch.libraries.storm.api.NodeSettings boltConfig,
                          List<org.thales.punch.libraries.objectstorage.tuples.ArchiveDevice> archiveDevices,
                          IFileNameSupplier fileNameSupplier,
                          org.thales.punch.libraries.objectstorage.tuples.ArchiveFormatSettings archiveFormatSettings)

Ctor. Extract fields from bolt config.

Parameters:

boltConfig - Bolt settings

archiveDevices - Url to archive devices

fileNameSupplier - Supplier for filename (Object storage or Simple filename)

archiveFormatSettings - Specific Batch settings

Method Detail

prepare

public void prepare(Map stormConf,
                    org.apache.storm.task.TopologyContext context,
                    org.apache.storm.task.OutputCollector outputCollector)

Prepare this bolt. Use this directly only for testing, and only after setting what you need in terms of configuration.

Specified by:

prepare in interface org.apache.storm.task.IBolt

Overrides:

prepare in class org.thales.punch.libraries.storm.api.BaseProcessingNode

flush

protected void flush(BatchArchiver batch)

Flush this batch, i.e. write its content to the configured archive(s). If required the batch metaData is forwarded as a tuple to be emitted further.

Parameters:

batch - batch to flush

initMetadata

protected org.thales.punch.libraries.objectstorage.indexing.BatchMetadata initMetadata(long timestamp)

Common metadata initialization

Parameters:

timestamp - beginning timestamp

Returns:

initialized metadata

getTimestampFromTuple

public long getTimestampFromTuple(org.apache.storm.tuple.ITuple tuple)

Extracts a timestamp from the received tuple.

Parameters:

tuple - the input tuple

Returns:

the associated timestamp, or the current time

addRuntimeData

public void addRuntimeData(Object context,
                           int key,
                           Object value)

Specified by:

addRuntimeData in interface org.thales.punch.libraries.punchlang.api.IRuntimeContext

getRuntimeData

public Object getRuntimeData(Object context,
                             int key)

Specified by:

getRuntimeData in interface org.thales.punch.libraries.punchlang.api.IRuntimeContext

createArchiver

protected BatchArchiver createArchiver(org.thales.punch.libraries.objectstorage.indexing.BatchMetadata metadata)

Create correct BatchArchiver from metadata

Parameters:

metadata - metadata

Returns:

BatchArchiver

getBloomFilterFields

public List<String> getBloomFilterFields()

Getter for bloom fields. (field must stay private for serialization)

Returns:

bloom fields