Skip to main content

BSPump is a real-time stream processor for Python 3.5+

Project description

Join the chat at https://gitter.im/TeskaLabs/bspump

Principles

  • Write once, use many times
  • Everything is a stream
  • Schema-less
  • Kappa architecture
  • Real-Time
  • High performance
  • Simple to use and well documented, so anyone can write their own stream processor
  • Asynchronous via Python 3.5+ async/await and asyncio
  • Event driven Architecture / Reactor pattern
  • Single-threaded core but compatible with threads
  • Compatible with pypy, Just-In-Time compiler capable of boosting Python code performace more then 5x times
  • Good citizen of the Python ecosystem
  • Modularized

Stream processor example

#!/usr/bin/env python3
import bspump
import bspump.socket
import bspump.common
import bspump.elasticsearch

class MyPipeline(bspump.Pipeline):
    def __init__(self, app):
        super().__init__(app)
        self.build(
            bspump.socket.TCPStreamSource(app, self),
            bspump.common.JSONParserProcessor(app, self),
            bspump.elasticsearch.ElasticSearchSink(app, self, "ESConnection")
        )


if __name__ == '__main__':
    app = bspump.BSPumpApplication()
    svc = app.get_service("bspump.PumpService")
    svc.add_connection(bspump.elasticsearch.ElasticSearchConnection(app, "ESConnection"))
    svc.add_pipeline(MyPipeline(app))
    app.run()

Video tutorial

http://img.youtube.com/vi/QvjiPxO4w6w/0.jpg

Blank application setup

You can clone blank application from it’s own repository.

Available technologies

  • bspump.amqp AMQP/RabbitMQ connection, source and sink
  • bspump.avro Apache Avro file source and sink
  • bspump.common Common processors and parsers
  • bspump.elasticsearch ElasticSearch connection, source and sink
  • bspump.file File sources and sinks (plain files, JSON, CSV)
  • bspump.filter Content, Attribute and TimeDrift filters
  • bspump.http.client HTTP client source, WebSocket client sink
  • bspump.http.web HTTP server source and sink, WebSocket server source
  • bspump.influxdb InfluxDB connection and sink
  • bspump.kafka Kafka connection, source and sink
  • bspump.mail SMTP connection and sink
  • bspump.mongodb MongoDB connection and lookup
  • bspump.mysql MySQL connection, source and sink
  • bspump.parquet Apache Parquet file sink
  • bspump.postgresql PostgreSQL connection and sink
  • bspump.slack Slack connection and sink
  • bspump.socket TCP source, UDP source
  • bspump.trigger Opportunistic, PubSub and Periodic triggers
  • bspump.crypto Cryptography
    • Hashing: SHA224, SHA256, SHA384, SHA512, SHA1, MD5, BLAKE2b, BLAKE2s
    • Symmetric Encryption: AES 128, AES 192, AES 256
  • bspump.analyzer
    • Time Window analyzer
    • Session analyzer
    • Geographical analyzer
    • Time Drift analyzer
  • bspump.lookup
    • GeoIP Lookup
  • bspump.unittest
    • Interface for testing Processors / Pipelines
  • bspump.oob Out-of-band sink and engine
  • bspump.web Pump API endpoints for pipelines, lookups etc.

Google Sheet with technological compatiblity matrix: https://docs.google.com/spreadsheets/d/1L1DvSuHuhKUyZ3FEFxqEKNpSoamPH2Z1ZaFuHyageoI/edit?usp=sharing

High-level architecture

Schema of BSPump high-level achitecture

Unit test

from unittest.mock import MagicMock
from bspump.unittest import ProcessorTestCase


class MyProcessorTestCase(ProcessorTestCase):

    def test_my_processor(self):

        # setup processor for test
        self.set_up_processor(my_project.processor.MyProcessor)

        # mock methods to suit your needs on pipeline ..
        self.Pipeline.method = MagicMock()

        # .. or instance of processor
        my_processor = self.Pipeline.locate_processor("MyProcessor")
        my_processor.method = MagicMock()

        output = self.execute(
            [(None, {'foo': 'bar'})]  # Context, event
        )

        # assert output
        self.assertEqual(
            [event for context, event in output],
            [{'FOO': 'BAR'}]
        )

        # asssert expected calls on `self.Pipeline.method` or `my_processor.method`
        my_processor.method.assert_called_with(**expected)

Running of unit tests

python3 -m unittest test

You can replace test with a location of your unit test module.

Licence

BSPump is an open-source software, available under BSD 3-Clause License.

Project details


Release history Release notifications

This version

19.7

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for bspump, version 19.7
Filename, size & hash File type Python version Upload date
bspump-19.7-py3-none-any.whl (138.0 kB) View hashes Wheel py3
bspump-19.7.tar.gz (83.8 kB) View hashes Source None

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page