The arrival and check in was quite normal with friendly ladies helping you charge money from your credit card. This part almost never fails, right? Me and my wife went to an elevator and put out baggage in the room and went to explore the city. The problems started when we came back.

When you just come into the room, there is a kitchen. That’s awesome, right? And then some details come into light. It is equipped with an exhaust hood (I’ve checked it actually works), but there is no cooking surface (wtf?). Looks like a grotesque joke to me. With only 2 glasses, 2 cups and a tiny water boiler. No tea or coffee are available on arrival. There are no knifes, no forks, nothing.

The next thing, the heating did not work. The heating was the largest problem for me, because my wife got cold easily. When I complained to reception…

Good luck on reaching out to the reception by the internal phone line. They do not pick up the phone or pick up and put you on hold. I had to come down from the 10th floor to talk to them. Twice.

So the reception, they had a standard procedure to go to the heating room and reset the heating system. Took them around 40 minutes to do so. After they realize that the problem is not solvable (I feel they were not surprised), they borrowed me an electric heater. Which I obviously can not use in the shower. Thus I should be ready to have wet air in the room for all duration of the stay. And yes of course, the heating problem is a regular issue. I spend twenty minutes reading bad reviews on Google maps for this hotel.

The screenshot is taken on 27.12.2018, a month ago would be in November 2018. Every second review is complaining about not working heating.

The building itself is built like an old Stalin’s panel house. When you expect this.

You should actually see this picture in your mind.

You can hear the farts of your neighbors. What is ridiculous is that they write to be quiet and do not play 'loud' music or disturb your neighbors by obnoxious behavior. I wonder if they consider farts to be obnoxious. I could hear people talking in the rooms nearby. The hotel was built from the ground up with no investment in the sound proofing. Just one more fact: our room was on the tenth floor and I could still hear the trains and the cars down there. Do not expect good sleep here.

And the house rules, omfg. Alarm! Somebody chew on your house rules. But wait, shouldn’t they have house rules with corporate style design and on good paper in a 4 star hotel? And "Tell your complaints to managers" they say. In our case, the management was simply not available. It was second day of Christmas, but it is a hotel. And it does not have anybody, who can take on responsibility to solve any problems. Yes, they can charge you for getting a room. No, they can not change a room in case of a problem.

I rant, rant, and rant again. What did I do to solve this? I called booking.com and in the end switched to another hotel. In the middle of the night, together with my wife. Trying to get my money back now.

The most relevant documentation for inotify is the inotify man page, that you can also read locally with:

$ man inotify

inotify has three parameters, that can be set in /proc interfaces: `max_user_instances`, `max_user_watches` and `max_queued_events`. We will look into each of them separately.

$ cat /proc/sys/fs/inotify/max_user_instances
128

This specifies an upper limit on the number of inotify instances that can be created per real user ID. inotify instance is created by inotify_init (2) call and returns a file descriptor referring to the inotify instance. Usually each process (e.g. tail) creates one inotify instance. There is an example down below.

$ cat /proc/sys/fs/inotify/max_user_watches
8192

This specifies an upper limit on the number of watches that can be created per real user ID. Each “watch” in the watch list specifies the pathname of a file or directory, along with some set of events that the kernel should monitor for the file referred to by that pathname. One inotify instance can have several inotify watches, which points to different directories/files.

$ cat /proc/sys/fs/inotify/max_queued_events
16384

`max_queued_events` is an upper limit on the number of events that can be queued to the corresponding inotify instance. As one inotify instance can have several “watches” (i.e. watching potentially large amount of directories/files), not all of the events (like file change, new file etc.) can be processed immediately. In this case, the events will populate an event queue, which has limit of 16384 events by default. the event queue can overflow. In this case, events are dropped (lost).

Now, I demonstrate how max_user_instances and max_user_watches work together. For the demo I will use the following commands:

$ export ILIMIT=128
$ for((i=0;i<=$ILIMIT;i+=1)); do echo $i >> $i.tmp; done
$ for((i=0;i<=$ILIMIT;i+=1)); do bash -c “tail -f $i.tmp &”; done
$ ps aux | grep -v grep | grep tail | awk ‘{print $2}’ | xargs kill -9

The first one is simply exports ILIMIT variable to your current shell and sets its’ value to 128. The second command creates ($ILIMIT+1) files in your current working directory with names 0.tmp, 1.tmp, 2.tmp etc. The third command starts tail -f ($ILIMIT + 1) times on all the files created by the previous command. Finally the last command look for all the tail processes and kills them.

Now, try to execute the commands with the $ILIMIT set to 128:

cd /tmp
export ILIMIT=128
for((i=0;i<=$ILIMIT;i+=1)); do echo $i >> $i.tmp; done
for((i=0;i<=$ILIMIT;i+=1)); do bash -c “tail -f $i.tmp &”; done

On Ubuntu 18.04 I hit the limit on 86th instance:

tail: inotify cannot be used, reverting to polling: Too many open files

I will now increase max_user_instances to set it higher than max_user_watches:

sudo sysctl -w fs.inotify.max_user_instances=10000
$ cat /proc/sys/fs/inotify/max_user_instances
10000

Clean up your previous tail commands:

ps aux | grep -v grep | grep tail | awk ‘{print $2}’ | xargs kill -9

Now try to create 8500 tail processes, which is lower than default max_user_watches value:

cd /tmp
export ILIMIT=8500
for((i=0;i<=$ILIMIT;i+=1)); do echo $i >> $i.tmp; done
for((i=0;i<=$ILIMIT;i+=1)); do bash -c “tail -f $i.tmp &”; done

I hit the limit on 7996 and have another inotify error message now:

tail: inotify resources exhausted
tail: inotify cannot be used, reverting to polling

Now let’s increase the max_user_watches and keep it in the same proportion as in default settings. Each inotify instance should have 64 watches available:

sudo sysctl -w fs.inotify.max_user_watches=640000
$ cat /proc/sys/fs/inotify/max_user_watches
640000

Clean up tail processes spawned previously:

ps aux | grep -v grep | grep tail | awk ‘{print $2}’ | xargs kill -9

Now let’s try it again:

cd /tmp
export ILIMIT=8500
for((i=0;i<=$ILIMIT;i+=1)); do echo $i >> $i.tmp; done
for((i=0;i<=$ILIMIT;i+=1)); do bash -c “tail -f $i.tmp &”; done

And now it has been successfully executed! To make a permanent change to the max_user_instances and max_user_watches variable, simply append these lines to /etc/sysctl.conf:

fs.inotify.max_user_instances=10000
fs.inotify.max_user_watches=640000

This will persist the changes after reboot of the machine. To apply the changes immediately, you need to reload the file:

$ sudo sysctl -p

I hope it clarifies a bit why this error happens and how to fix it.

• data.gov
• datahub.io
• opencanada
• publicdata.eu
• opendata.leipzig.de

To install ckan-aggregator-py, simply execute the following commands (you may need to install virtualenvwrapper before proceeding):

$ mkvirtualenv aggregator-py
$ cdvirtualenv
$ mkdir src && cd src
$ git clone git@github.com:dice-group/ckan-aggregator-py.git && cd ckan-aggregator-py
$ pip install -r requirements.txt

This will create Python virtual environment called aggregator-py and clone ckan-aggregator-py src/ckan-aggregator-py in the created virtual env.

Now, run init.sh script to create folders for storing crawled data:

$ ./init.sh

By default ckan-aggregator-py can access CKAN without the API key. If you want to have more requests to a CKAN endpoint than an average user, it is a good idea to register on a CKAN instance and use an API key. To configure an API key simply override __init__.py with __init__.py-example-key from the same folder. For example, for datahub.io:

$ cp ckanaggregatorpy/datahubio/__init__.py-example-key ckanaggregatorpy/datahubio/__init__.py

Then edit the new __init__.py file in your favorite editor.

We are now ready to crawl a CKAN datastore. Here we demonstrate usage of ckan-aggregator-py for scraping the Open Leipzig data portal. For this example, we implemented open-leipzig.py script. To run the example simply execute the script:

$ python open-leipzig.py

You will see output showing the number of packages in the CKAN instance, the number of RDF packages and example metadata of a single RDF package.

Leipzig Open Data portal contains 579 packages.
Updating the local package cache
...
...
...
Leipzig Open Data portal contains 85 packages, which contain RDF data
Showing metadata from CKAN for the first RDF package
{u'author': u'Amt f\xfcr Statistik und Wahlen',
u'author_email': u'mailto:statistik-wahlen@leipzig.de',
u'ckan_url': u'https://opendata.leipzig.de//dataset/allgemeinbildende-schulen',
u'creator_user_id': u'2046a73a-e0c2-42b6-bd8c-8da6ad8c552d',
...
...
...
u'version': None}

The metadata of the filtered packages can be further processed, transformed and loaded into applications (e.g. LODStats).

To read more blog posts on Data Science tools and libraries, subscribe to the blog of our DICE research group.

You might want to join our release webinar today (in under 4 hours). Registration and further details are available here: https://www.big-data-europe.eu/platform-launch-is-coming/

Spark Java applications have to be packaged into a .jar before it can be submitted to a Spark cluster. The packaging process is out of scope of this blog post, we will assume that you already have project with assembly tools (e.g. Maven or Gradle) set up and ready . For demonstration purposes we will use example application for Pi calculation from Spark Github repository:

$ mkdir SparkTutorial && cd SparkTutorial
$ git clone https://github.com/apache/spark && cd spark
$ build/mvn -DskipTests clean package

The “mvn clean package” will create an executable .jar file that is located in examples/target/ folder, which contains example applications. Spark project build takes a considerable amount of time and resources (~25 mins depending on your hardware) you might want to skip it. If you want to skip this step, you can download prebuilt .jar using the following commands:

$ mkdir examples/target
$ wget https://dl.dropboxusercontent.com/u/4882345/spark-examples/original-spark-examples_2.11-2.3.0-SNAPSHOT.jar -O examples/target/original-spark-examples_2.11-2.3.0-SNAPSHOT.jar

The next step is to start the Workbench. We will only start the parts of it that are necessary to run the application (i.e. only Hadoop and Spark without Hue/Spark-Notebook). You can download the most up to date docker-compose.yml for the Hadoop/Spark:

$ wget https://raw.githubusercontent.com/big-data-europe/docker-spark/2.1.0-hadoop2.8-hive-java8/docker-compose-minimal.yml -O docker-compose.yml

For the convenience we will create a Makefile containing the following:

You can download Makefile using the following command:

$ wget https://dl.dropboxusercontent.com/u/4882345/spark-examples/Makefile

Now we can start up the Workbench with “make up” command, stop it with “make down” and deploy our application with “make app” (we will create docker-compose-spark-app.yml in a moment). Start the Workbench:

$ make up

Check the workbench status by running:

$ docker ps

You should see 4 running containers: namenode, datanode, spark-master and spark-worker with status “Up xx seconds (healthy)”.

Now the Workbench is ready and we can pack and deploy the application. First create Dockerfile in the root of the project. We will pack .jar from examples/target/ folder:

You can copy/paste above snippet in Dockerfile or download our Dockerfile using the following command:

$ wget https://dl.dropboxusercontent.com/u/4882345/spark-examples/Dockerfile -O Dockerfile

The next step is to create docker-compose-spark-app.yml (also in the root of the project). It is very short docker-compose file, that simply runs the application in the same docker network as the Workbench:

Also, just copy/paste above snippet or simply run the following command to download our docker-compose file:

$ wget https://dl.dropboxusercontent.com/u/4882345/spark-examples/docker-compose-spark-app.yml

We are ready to run the application by executing the following command:

$ make app

In the execution logs you should see the result of execution:

app_1  | Pi is roughly 3.143235716178581

When you modify your application, you can easily rebuild and redeploy the project by using the following commands:

$ make build
$ make app

To summarize: we added the following files to the root of sample Maven project:

├── docker-compose-spark-app.yml
├── docker-compose.yml
├── Dockerfile
└── Makefile

We can start the Hadoop/Spark Workbench by executing:

$ make up

We can terminate Hadoop/Spark Workbench by executing:

$ make down

The application can be built and deployed by executing the following:

$ make build
$ make app

You can find a similar ready-made setup in docker-spark repository (see README.md).

For more information on BDE project refer to the project website, github org, twitter. Also, are also having a release webinar on 3rd of May (next week). You can register and participate here.

The complete example is located on my Github.

The basic idea behind the solution was not to modify any of the existing services, but to introduce a middleware, which will block users from making excessive amount of requests. As we are running most of our services in Docker containers, the solution should be able to run in the Docker. It is indeed possible with OpenRepose and the simplest setup inside docker-compose.yml looks like this:

version: “2.1”

services:
 repose-proxy:
 image: rackerlabs/repose:8.4.0.1
 volumes:
 — ./etc/repose:/etc/repose
 ports:
 — 80:80

nginx:
 image: nginx

Our actual service nginx is not exposed to the outside world and thus is hidden behind repose-proxy, which controls the flow of requests. The configuration of repose includes several things.

Main configuration file system-model.cfg.xml (located in ./etc/repose/system-model.cfg.xml) defines which port the Repose will listen to (port 80, which is bound to port 80 on outside host in docker-compose.yml, the binding in docker compose is always HOST:CONTAINER):

<node id=”repose_node1" hostname=”localhost” http-port=”80"/>

Defines two active filters (rate-limiting filter requires ip-user filter to inject X-PP-User HTTP header based on user IP address):

<filter name=”ip-user”/>
<filter name=”rate-limiting”/>

And the default proxy destination — nginx service (the hostname nginx should match the service name in docker-compose.yml definition and be in the same docker network as repose service):

<endpoint id=”service” protocol=”http” hostname=”nginx” root-path=”/” port=”80" default=”true”/>

The next thing is to configure IP ranges and assign them various user groups. In this example, we only define one user group match-all. The IP ranges configuration is ip-user.cfg.xml (located in ./etc/repose/ip-user.cfg.xml). We define match-all to match all IPv4 and IPv6:

<group name=”match-all”>
 <cidr-ip>0.0.0.0/0</cidr-ip>
 <cidr-ip>0::0/0</cidr-ip>
</group>

Now we can define the rate limit for the match-all group. The rate limit configuration is rate-limiting.cfg.xml (located in ./etc/repose/rate-limiting.cfg.xml). It uses the groups of users you defined in the IP ranges configuration previously (e.g. match-all group). Also, you can define the global limit for all the users:

<global-limit-group>
 <limit id=”global” uri=”*” uri-regex=”.*” value=”1000" unit=”MINUTE”/>
</global-limit-group>

The global limit affects the whole amount of requests to 1000 per minute. If this amount of requests is exceeded, no more requests will be possible. To limit requests for a particular group of users (i.e. match-all group) we define a limit group:

<limit-group id=”limited” groups=”match-all” default=”true”>
 <limit id=”all” uri=”*” uri-regex=”/.*” http-methods=”POST PUT GET DELETE” unit=”MINUTE” value=”10"/>
</limit-group>

This will limit requests for any given IP address within previously specified IP range (any IP addresses in this case) to 10 per one minute.

For additional information on OpenRepose refer to the official wiki and see examples (located in ./etc/repose/examples) folder for available filter configurations.

Thank you for reading, Ivan Ermilov :)

What I want from development setup is:

• It should be portable. I install it in a directory, then remove it at one point and forget about it. No apt-get, no packages.
• I can reinstall it any number of times in different directories on my PC. And it should not take much time (and no effort).
• All the binaries (hadoop, spark-submit etc.) should be available from a terminal session, but only if I work on a Spark project. Something similar to virtualenv in Python.
• A memo of the commands in one place, that can start pyspark shell or Hive beeline client.

What I created was a Makefile to install everything into one folder. It is available on the Github. For the full installation instructions, please read README.md, I will put the commands here which you need to run to install the environment:

mkdir -p ~/Workspace/hadoop-spark-hive && cd ~/Workspace/hadoop-spark-hive
git clone https://github.com/earthquakesan/hdfs-spark-hive-dev-setup ./
make download
make configure

And then you can run start Hadoop/Spark with:

make start_hadoop
make start_spark

Check Hadoop at http://localhost:50070 (one datanode should be available), Spark at http://localhost:8080 (one spark worker should be available) and Hive with:

#Hive configuration requires Hadoop/HDFS running
make configure_hive
make start_hive_server
#In another terminal start beeline client
make start_hive_beeline_client

Then throw these queries into the client:

CREATE TABLE pokes (foo INT, bar STRING);
LOAD DATA LOCAL INPATH './tools/apache-hive-2.1.0-bin/examples/files/kv1.txt' OVERWRITE INTO TABLE pokes;
DESCRIBE pokes;
SELECT * FROM pokes;

If you’ve got no error messages, congratulations — it works!

Ok, now the fun part. To start Python interactive shell:

make pyspark

To start Scala interactive shell:

make spark_shell

To export Hadoop/Spark/Hive binaries to your PATH:

make activate
source activate

Now you can access the HDFS easily:

➜  hdfs dfs -ls /
Found 2 items
drwxrwxr-x   - ivan supergroup          0 2016-07-20 12:08 /tmp
drwxr-xr-x   - ivan supergroup          0 2016-07-20 16:16 /user

Or do spark-submit for your application.

Submitting example Spark application to your HDFS/Spark/Hive setup

As an example, I will use an application from O’Reilly “Learning Spark” (I recommend this book, it is written by Spark devs and provides a lot of useful insights). The code is available from the databricks github.

cd ~/Workspace/hadoop-spark-hive
mkdir src && cd src
git clone https://github.com/databricks/learning-spark
cd learning-spark/mini-complete-example

Now, you have to change build.sbt to match our Spark version to look like this (we use Spark “1.6.2”, spark-core should be the same version):

➜ cat build.sbt 
name := "learning-spark-mini-example"

version := "0.0.1"

scalaVersion := "2.10.4"

// additional libraries
libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "1.6.2" % "provided"
)

Now can build and submit the app to Spark:

sbt clean package
spark-submit --class com.oreilly.learningsparkexamples.mini.scala.WordCount
./target/scala-2.10/learning-spark-mini-example_2.10-0.0.1.jar
file://$(pwd)/README.md file://$(pwd)/wordcounts

It will read the file from your current working directory (the project root) and write the results to wordcounts/ folder. If you will not use file:// prefix, Spark will read/write to HDFS.

That’s it! If you have any questions or something did not work, please open an issue on github. Also, for more cool stuff on Big Data, visit Big Data Europe project blog.

In this article I discuss how I created scalable HDFS/Spark setup using docker and docker-compose.

So, why do I need to pack Spark into docker in the first place? We have a cluster with three servers, each has 256 GB RAM and 64 CPU cores. As we run some other applications on this cluster, we have chosen Docker to isolate applications from each other. Yet being able to connect them when necessary. Also, docker can restrict RAM and set up CPU quota for each container. Thus, in theory you can spawn several spark-workers isolated from each other on the same machine. For instance, you can spawn 60 spark-workers on one of those servers with 2 GB RAM and 1/70 of CPU time each.

Docker exists for a couple of years already and has an ecosystem, where users can find dozens of docker images for almost all of mainstream technologies. Including HDFS/Spark. So I collected all the available images by searching for “spark” in the Docker Hub looking for something I can use out of the box. I put all of those images into imagelayers and got a nice visualization. Then I shortlisted them by the following criteria:

• Less than 1 GB in size. The bigger images most likely do not have separation of components, which makes them hard to scalable.
• HDFS/Spark separated? I want to scale HDFS nodes separately from Spark nodes.
• Is it easy to install? Is there documentation?
• Can I scale Spark with this image?
• Is it easy to connect Spark to HDFS?

After checking against criteria 1 and 2, which are relatively easy to figure out without trying images out, I had to run all of them one by one. And I found out that none of the images fit criterion 4.

If I throw out scaling, uhopper/hadoop-spark looks good. So, I forked it and modified for the setup. Let’s see how to use my simplified version! (use git to clone my repo).

$ ./buildall.sh
$ docker network create hadoop
$ docker-compose up

That’s it! buildall.sh will iterate through all the Dockerfiles inside the repo and build necessary images. As I am using version 1 of docker-compose, you’ll have to create docker network manually.

You have 1 namenode, 2 datanodes, 1 spark master and 1 spark worker as well as spark-notebook running. Navigate to http://your.docker.host:50070 to see status of your HDFS and to http://your.docker.host:8080 for Spark.

On Spark page you will only see 1 worker. Let’s scale it up to 5.

$ docker-compose scale spark-worker=5

Refresh the page on http://your.docker.host:8080. You will see 5 spark-workers registered now.

Let’s throw some data into our HDFS now and run a simple job. Let’s use Waterworks — Treatment csv dataset from the EU data portal.

<-- you should be inside gitrepo folder (and run docker-compose, data dir would be created automatically)
$ cd data
$ wget https://data.mattilsynet.no/vannverk/vannbehandlingsanlegg.csv
$ docker run -it --rm --env-file=../hadoop.env --net hadoop uhopper/hadoop hadoop fs -mkdir -p /user/root
$ docker run -it --rm --env-file=../hadoop.env --volume $(pwd):/data --net hadoop uhopper/hadoop hadoop fs -put /data/vannbehandlingsanlegg.csv /user/root

Go to http://localhost:50070 and check if file exists under the path ‘/user/root/vannbehandlingsanlegg.csv’.

Open spark-notebook on localhost:9000 and choose “core/simple spark” example. Run the cells by clicking on them and pushing play button. Then add new cell (using a plus button in the toolbar). Let’s read our file:

val textFile = sc.textFile(“/user/root/vannbehandlingsanlegg.csv”)
textFile.count()

It will show you the execution time and the result as such:

textFile: org.apache.spark.rdd.RDD[String] = /user/root/vannbehandlingsanlegg.csv MapPartitionsRDD[2] at textFile at <console>:53
res8: Long = 4071
4071

That’s it! If you have any questions, feel free to open new issues on github, or leave a comment to this article.