PipelineDB: working with data streams / Selectel Blog

In previous publications, we have already addressed the problem of real-time event processing. Today we would like to return to this topic again and talk about a new and very interesting tool - the streaming DBMS PipelineDB .


PipelineDB is based on the PostgreSQL 9.4 code base and is fully compatible with it. Its first release took place in July 2015, and in January 2016 the enterprise version was released .

Below we will compare PipelineDB with existing solutions of a similar plan, we will give a brief installation and initial setup instructions, as well as a practical example.

Real-time data processing: an excursion into history

The principle of operation of PipelineDB can be formulated as follows: "constant requests, short-term data." In relational DBMS, the situation is exactly the opposite: “short-term queries, persistent data. In PipelineDB, data is not stored, but is received in a stream; their processing takes place "on the fly", in motion.

The first attempts to create tools for processing data in motion date back to the end of the 1980s, when the so-called Active Database Systems appeared . They were extensions to existing DBMSs for event processing using triggers and rules . Examples of such solutions include HiPAC , Starburst, or ODE .
However, they did not get wide distribution: the scope of their application was rather narrow, and the syntax of the rules was too complicated and confusing.

In the 1990s and early 2000s, Data Stream Management Systems appeared : TelegraphCQ (fork of PostgreSQL), StreamBase , StreamSQL . The principle of operation of these tools was as follows: using the so-called window operators (window operators), the flows were converted into tables, in relation to which it was then possible to apply SQL queries.
The appearance of such solutions was an undoubted step forward, but they could not provide high speed and performance when working with large data streams.

Tools oriented to data processing without storage have gained distribution over the past 5-6 years. Of the most famous examples, it is worth highlighting, in particular, Storm and Heron . Of the relatively recent ones - Apache Calcite . All these solutions are characterized by the complexity of installation and configuration, as well as a very high entry threshold.

The advantages of PipelineDB over the tools mentioned above are obvious:

• ease of setup: to get started, just download and install the necessary packages;
• ease of development (due to compatibility with PostgreSQL).

Let's consider how work with data streams is built in PipelineDB. Let's start with an analysis of the two most important concepts: “continuous presentation” and “flow”.

Streams and Continuous Views

“Stream” and “continuous presentation” are the main abstractions of PipelineDB.
A stream  is a sequence of events. Writing events to a stream is carried out in exactly the same way as writing to tables in relational DBs (for more details, see here ). When an event arrives in a stream, a timestamp is added to it.

Streams in PipelineDB have an auxiliary function, which is to supply data for continuous presentations. Unlike tables, flows do not need to create schemas. You can write data to a stream while it interacts with at least one continuous view.

Continuous performance(Eng. continuous view) is a selection of streams and tables, updated as new data arrives. In continuous representations fall events selected by certain parameters.

To better understand how PipelineDB works, here are some examples of continuous views.

So, for example, you can create a continuous view for daily counting the number of unique visitors coming to the site via external links:

CREATE CONTINUOUS VIEW uniques AS
SELECT date_trunc('day', arrival_timestamp) AS day,
  referrer::text, COUNT(DISTINCT user_id::integer)
FROM users_stream GROUP BY day, referrer;

Another example is counting the number of times an ad has been displayed on a site in the last 5 minutes:

CREATE CONTINUOUS VIEW imps AS
SELECT COUNT(*) FROM imps_stream
WHERE (arrival_timestamp > clock_timestamp() - interval '5 minutes');

As you can see, continuous representations have the following form:

CREATE CONTINUOUS VIEW name AS query

When creating a continuous view in relation to threads, the SELECT operation is performed; with its help, data corresponding to the required parameters is selected.

The main theoretical information necessary for understanding the principles of PipelineDB, we have outlined. We pass to the practical part. First, we describe the installation and initial configuration of PipelineDB, and then move on to a practical example.

Installation and initial setup

We will describe the installation procedure for PipelineDB on OC Ubuntu 14.04. If you are using a different Linux distribution, consult the  official documentation .

To install PipelineDB, just run two commands:

$ wget https://www.pipelinedb.com/download/0.9.3/ubuntu14
$ sudo dpkg -i ubuntu14

After that, we initialize the PipelineDB server:

$ pipeline-init -D [имя директории]

In the -D option, you can specify the name of a new directory that will be created automatically. Here is a list of the contents of this directory:

base          pg_hba.conf    pg_replslot   pg_subtrans  pipelinedb.auto.conf
global        pg_ident.conf  pg_serial     pg_tblspc    pipelinedb.conf
pg_clog       pg_logical     pg_snapshots  pg_twophase  postmaster.opts
pg_commit_ts  pg_multixact   pg_stat       PG_VERSION   postmaster.pid
pg_dynshmem   pg_notify      pg_stat_tmp   pg_xlog

The main settings of PipelineDB are stored in the pipelinedb.conf file. They hardly differ from the corresponding PostgreSQL settings.

By default, PipelineDB cannot accept connections from remote hosts. To change this setting, open the pipelinedb.conf file, find the Connections and Authentication section in it, uncomment the first line and edit it as follows:

listen_addresses = '*'

After that, we will write the specific hosts in the pg_hba.conf file:

host    all             all             /<подсеть>            md5

If we need to accept connections from all possible hosts, this line should look like this:

host    all             all             0.0.0.0/0            md5

That's all. PipelineDB is ready to go.
To run it in the background, execute the following command:

$ pipeline-ctl -D [имя директории] -l pipelinedb.log start

Case Study: Analyzing Wikipedia Statistics

We examined the necessary theory, and also described the installation and initial configuration of PipelineDB. We turn to the features of using PipelineDB in practice.

We will consider an interesting example, which is given in the official documentation of PipelineDB: analysis of statistics on visits to Wikipedia pages and related projects per hour (Wiktionary, Wikisources, Wikibooks and others). These statistics are available in the public domain . Information about each appeal is presented in the form of a record consisting of the following fields:

circulation time | project | number of views per | total byte served

We will be interested in the maximum, minimum and average number of hits per page for an hour, as well as the 99th percentile of hits.
We activate the execution of continuous requests:

$ psql -h localhost -p 5432 -d pipeline -c "ACTIVATE"

After that, create a continuous view:

$ psql -h localhost -p 5432 -d pipeline -c "CREATE CONTINUOUS VIEW wiki_stats AS
SELECT hour::timestamp, project::text,
           count(*) AS total_pages,
           sum(view_count::bigint) AS total_views,
           min(view_count) AS min_views,
           max(view_count) AS max_views,
           avg(view_count) AS avg_views,
           percentile_cont(0.99) WITHIN GROUP (ORDER BY view_count) AS p99_views,
           sum(size::bigint) AS total_bytes_served
    FROM wiki_stream
    GROUP BY hour, project;"
CREATE CONTINUOUS VIEW

The above command indicates that we will receive data for continuous presentation from the wiki_stream. To create such a stream, we need to download data from the site, unzip it, write it to standard output, and then transfer PipelineDB using the COPY command:

$ curl -sL http://pipelinedb.com/data/wiki-pagecounts | gunzip | \
        psql -h localhost -p 5432 -d pipeline -c "
        COPY wiki_stream (hour, project, title, view_count, size) FROM STDIN"

Note that the amount of data is very large (they are stored in the form of archives of 80-90 MB each), and downloading them can take a long time. You can stop the download at any time by pressing the standard key combination Ctrl + C.

When the download is complete, execute the command:

$ psql -h localhost -p 5432 -d pipeline -c "
SELECT * FROM wiki_stats ORDER BY total_views DESC";

The result will be presented in the form of a table (we give only a small fragment):

Conclusion

PipelineDB is an interesting and promising product. We hope that it will continue to develop successfully.
If you have experience using PipelineDB in practice, we will be glad if you share your experience in the comments.

For those who want to know more, here are some useful links:

• PipelineDB official documentation ;
• Blog PipelineDB developers ;
• data analysis with PipelineDB: a practical case ;
• PipelineDB in a container: a practical case .

Anyone who for one reason or another can not leave comments here, we invite you to our blog .