Blog

Consuming from nsq to Hadoop HDFS with Apache Flume

10/25/2018 by Luc

I was working on a prototype for consuming from nsq to the Hadoop HDFS file system with Apache Flume. I couldn’t find much information, so I made a small project.

Below are a few extra notes about nsq and the project.

docker-flume/Dockerfile builds a container which includes Flume, nsq and Hadoop core necessary for persisting files to HDFS
In order to be able to quickly modify some files in the built docker container, some extra paths are mapped in the docker-compose.yml:
- Adding the start-flume.sh script allows updating the classpath to quickly add libraries and restart the container.
- Adding the flume/flume.conf path means the flume config can be adjusted and the container restarted, without requiring a container rebuild.
In flume.conf docker.sources.r1.command specifies nsq_tail. While this is a quick way to consume from nsq, it is possibly not a reliable technique to use in a production environment.
nsq doesn’t appear to have a mechanism for resuming from an offset, like Kafka. This means there could be gaps in data if a consumer goes down and comes back up. In this sense, nsq has similarities to high volume messaging systems like RabbitMQ or ZeroMQ, which provide fewer message durability guarantees than Kafka.

Pyenv Quick Reference

12/04/2017 by Luc

Install Pyenv, pyenv-virtualenv and pyenv-virtualenvwrapper
- Pyenv Installer
Usage
- List Available Python Versions
Install a specific version
- Create A New virtualenv Based on a Specific Python Version
Customize pip.conf
Other Useful Commands

Pyenv is a tool for easily switching between multiple versions of Python. One way to use the tool is in conjunction with the pyenv-virtualenv and pyenv-virtualenvwrapper plugins.

Detailed instructions are available here and here. The following reference is a brief summary of some frequently used commands.

Install Pyenv, pyenv-virtualenv and pyenv-virtualenvwrapper

Pyenv Installer

First use the Pyenv Installer. The ‘Github Way’ of installing is easiest.

Edit your profile (i.e. ~/.bash_profile, ~/.zshenv, etc) and add the following, adjusting paths as necessary:

# Use pyenv for Python version management
export PYENV_ROOT="$HOME/Dev/.pyenv" <--Use your chosen location here
export PATH="$PYENV_ROOT/bin:$PATH"
eval "$(pyenv init -)"

# Use pyenv for Python version management

export PYENV_ROOT="$HOME/Dev/.pyenv" <--Use your chosen location here

export PATH="$PYENV_ROOT/bin:$PATH"

eval "$(pyenv init -)"

Then install the following two plugins:
– pyenv-virtualenv
– pyenv-virtualenvwrapper

Edit your profile again, adding the following:

eval "$(pyenv virtualenv-init -)"
export WORKON_HOME=$HOME/.virtualenvs
export PROJECT_HOME=$HOME/Dev
source /usr/local/bin/virtualenvwrapper.sh
pyenv virtualenvwrapper

eval "$(pyenv virtualenv-init -)"

export WORKON_HOME=$HOME/.virtualenvs

export PROJECT_HOME=$HOME/Dev

source /usr/local/bin/virtualenvwrapper.sh

pyenv virtualenvwrapper

Usage

List Available Python Versions

pyenv install -l

1 2	pyenv install -l

Install a specific version

pyenv install 3.6.4

1 2	pyenv install 3.6.4

Create A New virtualenv Based on a Specific Python Version

See the full guide here.

pyenv versions
pyenv shell 3.6.4
mkvirtualenv --python=`pyenv which python` mynewenv

pyenv versions

pyenv shell 3.6.4

mkvirtualenv --python=`pyenv which python` mynewenv

Customize pip.conf

You can set a project or virtualenv specific pip.conf by creating one in
$VIRTUAL_ENV/pip.conf when the virtualenv is active:

$ workon myenv
$ vi $VIRTUAL_ENV/pip.conf

$ workon myenv

$ vi $VIRTUAL_ENV/pip.conf

Add this:

[global]
index-url = https://<path_to_a_corporate_repo>

[global]

index-url = https://<path_to_a_corporate_repo>

Other Useful Commands

Command	Description
lsvirtualenv	List all the virtualenvs
echo $WORKON_HOME	Show the directory where virtualenvs are located
workon myvirtualenv	Set the current virtualenv to myvirtualenv
pyenv global	Show the global Python version currently in use
pyenv versions	Show which Python versions are installed
pyenv global 3.6.4	Switch to the 3.6.4 version if installed

Kafkacat Quick Reference

08/23/2017 by Luc

Produce
Consume
List Topics
Read Messages From File
Generate a Test File of Messages
Read the Last n Messages from Partition n

kafkacat is a useful tool for working with Kafka brokers. It’s quick and very simple to use. This is a cheatsheet of some useful commands.

Produce

$ echo "foo" | kafkacat -b localhost:9092 -t mytopic

1 2	$ echo "foo" \| kafkacat -b localhost:9092 -t mytopic

Consume

kafkacat -C -b localhost:9092  -t mytopic

1 2	kafkacat -C -b localhost:9092 -t mytopic

List Topics

kafkacat  -b localhost:9092 -L -t mytopic

1 2	kafkacat -b localhost:9092 -L -t mytopic

Read Messages From File

Read messages from a file, with each message separated by a newline:

kafkacat -P -l -b kafka:9092 -t mytopic myfile

1 2	kafkacat -P -l -b kafka:9092 -t mytopic myfile

Generate a Test File of Messages

Given a file containing a DATE variable, generate a set of messages with incrementing timestamp, for use with the previous example:

for i in {1..5000}; do DATE=`date -u +%d/%b/%Y:%H:%M:%S`; sed "s~DATE~$DATE~" template.msg >> myfile; done

1 2	for i in {1..5000}; do DATE=`date -u +%d/%b/%Y:%H:%M:%S`; sed "s~DATE~$DATE~" template.msg >> myfile; done

Read the Last n Messages from Partition n

This example uses the -o flag to read from a specified offset. Specifying -3 allows retrieval of the last 3 messages. The -p option targets partition 0:

kafkacat -C -b kafka -t mytopic -o -3 -e -p 0

1 2	kafkacat -C -b kafka -t mytopic -o -3 -e -p 0

How to Set Up Kafka on Hortonworks Sandbox (Docker version)

03/24/2017 by Luc

Assumes a container was previously set up following these instructions and you now want to enable the Kafka service.

Stop and delete existing an existing sandbox container if one was already created:

docker stop sandbox docker rm sandbox

1
2
3

docker stop sandbox
docker rm sandbox
Add the default Hortonworks Kafka port to the create_container.sh script:

-p 6667:6667

1
2

-p 6667:6667
Recreate and start the container:

./create_container.sh

1
2

./create_container.sh

SSH to the new container and start services. The password for the newly created container will have been reset to the hortonworks default (hadoop), so you’ll need to reset it at this point:

ssh -p 2222 root@localhost
The password for the newly created container will have been reset to the hortonworks default (`hadoop`), so you'll need to reset it at this point.

ssh -p 2222 root@localhost

The password for the newly created container will have been reset to the hortonworks default (`hadoop`), so you'll need to reset it at this point.

When started, go to http://localhost:8080/#/login login as maria_dev/ maria_dev.
Start the Kafka service from the menu.

Check it works (assumes a local Kafka installation on your development machine).

In one terminal session:

cd $KAFKA_HOME bin/kafka-console-producer.sh --topic test --broker-list sandbox.hortonworks.com:6667 (Type a message)

1
2
3
4

cd $KAFKA_HOME
bin/kafka-console-producer.sh --topic test --broker-list sandbox.hortonworks.com:6667
(Type a message)

In a second session:

cd $KAFKA_HOME
./kafka-console-consumer.sh --bootstrap-server localhost:6667 --topic test --from-beginning --zookeeper localhost:2181
(Check the message from the first session reaches the second session)

cd $KAFKA_HOME

./kafka-console-consumer.sh --bootstrap-server localhost:6667 --topic test --from-beginning --zookeeper localhost:2181

(Check the message from the first session reaches the second session)

PostgreSQL Quick Reference

01/01/2017 by Luc

Connect

$ psql -h localhost -d mydb -U myuser

1 2	$ psql -h localhost -d mydb -U myuser

List Databases

From within a psql session:

l

List Tables

dt

Connect To DB

c myuser

c myuser

Execute SQL in a File

$ psql -U myuser -d mydb -a -f myfile.sql

1 2	$ psql -U myuser -d mydb -a -f myfile.sql

Run a Query

$ psql -U myuser -d mydb -c 'SELECT * FROM my_table'

1 2	$ psql -U myuser -d mydb -c 'SELECT * FROM my_table'

Output to File

Example shows opening a file, listing tables to the file, and finally closing the file:

db=>o out.txt
db=>dt
db=>q

db=>o out.txt

db=>dt

db=>q

Get PostgreSQL Logs

db=> show log_destination;

1 2	db=> show log_destination;

See this reference.

Check Running Queries

SELECT pid, age(query_start, clock_timestamp()), usename, query 
FROM pg_stat_activity 
WHERE  query NOT ILIKE '%pg_stat_activity%' 
ORDER BY query_start desc;

SELECT pid, age(query_start, clock_timestamp()), usename, query

FROM pg_stat_activity

WHERE query NOT ILIKE '%pg_stat_activity%'

ORDER BY query_start desc;

Terminate a Running Query

select pg_terminate_backend(pid int)

1 2	select pg_terminate_backend(pid int)

Check Locks

See this reference.

Do Something for Every Child Table

DO
$$
DECLARE
    row record;
BEGIN
    FOR row IN select c.relname as tablename from pg_inherits JOIN pg_class AS c ON (inhrelid=c.oid) JOIN pg_class AS p ON (inhparent=p.oid) WHERE p.relname = 'my_table'
    LOOP
        EXECUTE 'drop TABLE ' || quote_ident(row.tablename);
    END LOOP;
END;
$$;

DECLARE

row record;

BEGIN

FOR row IN select c.relname as tablename from pg_inherits JOIN pg_class AS c ON (inhrelid=c.oid) JOIN pg_class AS p ON (inhparent=p.oid) WHERE p.relname = 'my_table'

LOOP

EXECUTE 'drop TABLE ' || quote_ident(row.tablename);

END LOOP;

END;

$$;

Insert Sample Data with Incrementing Values

insert into my_table (name, thing, id)
select 'user1', 'test' || x.id, 1
    from generate_series(40, 100, 1) as x(id);

insert into my_table (name, thing, id)

select 'user1', 'test' || x.id, 1

from generate_series(40, 100, 1) as x(id);

Check the generate_series documentation for what the parameters mean.

Convert a String to a Timestamp

to_timestamp('2015-04-30 23:59:59', 'YYYY-MM-DD HH24:MI:SS')

1 2	to_timestamp('2015-04-30 23:59:59', 'YYYY-MM-DD HH24:MI:SS')

Partitioning

Refer to this example for how to partition tables for performance.