Martin Krasser's Bloghttp://krasserm.github.io/
Tue, 13 Feb 2018 16:12:08 +0000Tue, 13 Feb 2018 16:12:08 +0000Jekyll v3.6.2Deep face recognition with Keras, Dlib and OpenCV<p><strong>There is also <a href="http://nbviewer.jupyter.org/github/krasserm/face-recognition/blob/master/face-recognition.ipynb?flush_cache=true">companion notebook</a> for this article on <a href="https://github.com/krasserm/face-recognition">Github</a>.</strong></p>
<p>Face recognition identifies persons on face images or video frames. In a nutshell, a face recognition system extracts features from an input face image and compares them to the features of labeled faces in a database. Comparison is based on a feature similarity metric and the label of the most similar database entry is used to label the input image. If the similarity value is above a certain threshold the input image is labeled as <em>unknown</em>. Comparing two face images to determine if they show the same person is known as face verification.</p>
<p>This article uses a deep convolutional neural network (CNN) to extract features from input images. It follows the approach described in <a href="https://arxiv.org/abs/1503.03832">[1]</a> with modifications inspired by the <a href="http://cmusatyalab.github.io/openface/">OpenFace</a> project. <a href="https://keras.io/">Keras</a> is used for implementing the CNN, <a href="http://dlib.net/">Dlib</a> and <a href="https://opencv.org/">OpenCV</a> for aligning faces on input images. Face recognition performance is evaluated on a small subset of the <a href="http://vis-www.cs.umass.edu/lfw/">LFW</a> dataset which you can replace with your own custom dataset e.g. with images of your family and friends if you want to further experiment with the <a href="https://github.com/krasserm/face-recognition">notebook</a>. After an overview of the CNN architecure and how the model can be trained, it is demonstrated how to:</p>
<ul>
<li>Detect, transform, and crop faces on input images. This ensures that faces are aligned before feeding them into the CNN. This preprocessing step is very important for the performance of the neural network.</li>
<li>Use the CNN to extract 128-dimensional representations, or <em>embeddings</em>, of faces from the aligned input images. In embedding space, Euclidean distance directly corresponds to face similarity.</li>
<li>Compare input embedding vectors to labeled embedding vectors in a database. Here, a support vector machine (SVM) and a KNN classifier, trained on labeled embedding vectors, play the role of a database. Face recognition in this context means using these classifiers to predict the labels i.e. identities of new inputs.</li>
</ul>
<h2 id="cnn-architecture-and-training">CNN architecture and training</h2>
<p>The CNN architecture used here is a variant of the inception architecture <a href="https://arxiv.org/abs/1409.4842">[2]</a>. More precisely, it is a variant of the NN4 architecture described in <a href="https://arxiv.org/abs/1503.03832">[1]</a> and identified as <a href="https://cmusatyalab.github.io/openface/models-and-accuracies/#model-definitions">nn4.small2</a> model in the OpenFace project. This article uses a Keras implementation of that model whose definition was taken from the <a href="https://github.com/iwantooxxoox/Keras-OpenFace">Keras-OpenFace</a> project. The architecture details aren’t too important here, it’s only useful to know that there is a fully connected layer with 128 hidden units followed by an L2 normalization layer on top of the convolutional base. These two top layers are referred to as the <em>embedding layer</em> from which the 128-dimensional embedding vectors can be obtained. The complete model is defined in <a href="https://github.com/krasserm/face-recognition/blob/master/model.py">model.py</a> and a graphical overview is given in <a href="https://github.com/krasserm/face-recognition/blob/master/model.png">model.png</a>. A Keras version of the nn4.small2 model can be created with <code class="highlighter-rouge">create_model()</code>.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">from</span> <span class="nn">model</span> <span class="kn">import</span> <span class="n">create_model</span>
<span class="n">nn4_small2</span> <span class="o">=</span> <span class="n">create_model</span><span class="p">()</span>
</code></pre></div></div>
<p>Model training aims to learn an embedding <script type="math/tex">f(x)</script> of image <script type="math/tex">x</script> such that the squared L2 distance between all faces of the same identity is small and the distance between a pair of faces from different identities is large. This can be achieved with a <em>triplet loss</em> <script type="math/tex">L</script> that is minimized when the distance between an anchor image <script type="math/tex">x^a_i</script> and a positive image <script type="math/tex">x^p_i</script> (same identity) in embedding space is smaller than the distance between that anchor image and a negative image <script type="math/tex">x^n_i</script> (different identity) by at least a margin <script type="math/tex">\alpha</script>.</p>
<script type="math/tex; mode=display">L = \sum^{m}_{i=1} \large[ \small {\mid \mid f(x_{i}^{a}) - f(x_{i}^{p})) \mid \mid_2^2} - {\mid \mid f(x_{i}^{a}) - f(x_{i}^{n})) \mid \mid_2^2} + \alpha \large ] \small_+</script>
<p><script type="math/tex">[z]_+</script> means <script type="math/tex">max(z,0)</script> and <script type="math/tex">m</script> is the number of triplets in the training set. The triplet loss in Keras is best implemented with a custom layer as the loss function doesn’t follow the usual <code class="highlighter-rouge">loss(input, target)</code> pattern. This layer calls <code class="highlighter-rouge">self.add_loss</code> to install the triplet loss:</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">from</span> <span class="nn">keras</span> <span class="kn">import</span> <span class="n">backend</span> <span class="k">as</span> <span class="n">K</span>
<span class="kn">from</span> <span class="nn">keras.models</span> <span class="kn">import</span> <span class="n">Model</span>
<span class="kn">from</span> <span class="nn">keras.layers</span> <span class="kn">import</span> <span class="n">Input</span><span class="p">,</span> <span class="n">Layer</span>
<span class="c"># Input for anchor, positive and negative images</span>
<span class="n">in_a</span> <span class="o">=</span> <span class="n">Input</span><span class="p">(</span><span class="n">shape</span><span class="o">=</span><span class="p">(</span><span class="mi">96</span><span class="p">,</span> <span class="mi">96</span><span class="p">,</span> <span class="mi">3</span><span class="p">))</span>
<span class="n">in_p</span> <span class="o">=</span> <span class="n">Input</span><span class="p">(</span><span class="n">shape</span><span class="o">=</span><span class="p">(</span><span class="mi">96</span><span class="p">,</span> <span class="mi">96</span><span class="p">,</span> <span class="mi">3</span><span class="p">))</span>
<span class="n">in_n</span> <span class="o">=</span> <span class="n">Input</span><span class="p">(</span><span class="n">shape</span><span class="o">=</span><span class="p">(</span><span class="mi">96</span><span class="p">,</span> <span class="mi">96</span><span class="p">,</span> <span class="mi">3</span><span class="p">))</span>
<span class="c"># Output for anchor, positive and negative embedding vectors</span>
<span class="c"># The nn4_small model instance is shared (Siamese network)</span>
<span class="n">emb_a</span> <span class="o">=</span> <span class="n">nn4_small2</span><span class="p">(</span><span class="n">in_a</span><span class="p">)</span>
<span class="n">emb_p</span> <span class="o">=</span> <span class="n">nn4_small2</span><span class="p">(</span><span class="n">in_p</span><span class="p">)</span>
<span class="n">emb_n</span> <span class="o">=</span> <span class="n">nn4_small2</span><span class="p">(</span><span class="n">in_n</span><span class="p">)</span>
<span class="k">class</span> <span class="nc">TripletLossLayer</span><span class="p">(</span><span class="n">Layer</span><span class="p">):</span>
<span class="k">def</span> <span class="nf">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">alpha</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">):</span>
<span class="bp">self</span><span class="o">.</span><span class="n">alpha</span> <span class="o">=</span> <span class="n">alpha</span>
<span class="nb">super</span><span class="p">(</span><span class="n">TripletLossLayer</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="n">__init__</span><span class="p">(</span><span class="o">**</span><span class="n">kwargs</span><span class="p">)</span>
<span class="k">def</span> <span class="nf">triplet_loss</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">inputs</span><span class="p">):</span>
<span class="n">a</span><span class="p">,</span> <span class="n">p</span><span class="p">,</span> <span class="n">n</span> <span class="o">=</span> <span class="n">inputs</span>
<span class="n">p_dist</span> <span class="o">=</span> <span class="n">K</span><span class="o">.</span><span class="nb">sum</span><span class="p">(</span><span class="n">K</span><span class="o">.</span><span class="n">square</span><span class="p">(</span><span class="n">a</span><span class="o">-</span><span class="n">p</span><span class="p">),</span> <span class="n">axis</span><span class="o">=-</span><span class="mi">1</span><span class="p">)</span>
<span class="n">n_dist</span> <span class="o">=</span> <span class="n">K</span><span class="o">.</span><span class="nb">sum</span><span class="p">(</span><span class="n">K</span><span class="o">.</span><span class="n">square</span><span class="p">(</span><span class="n">a</span><span class="o">-</span><span class="n">n</span><span class="p">),</span> <span class="n">axis</span><span class="o">=-</span><span class="mi">1</span><span class="p">)</span>
<span class="k">return</span> <span class="n">K</span><span class="o">.</span><span class="nb">sum</span><span class="p">(</span><span class="n">K</span><span class="o">.</span><span class="n">maximum</span><span class="p">(</span><span class="n">p_dist</span> <span class="o">-</span> <span class="n">n_dist</span> <span class="o">+</span> <span class="bp">self</span><span class="o">.</span><span class="n">alpha</span><span class="p">,</span> <span class="mi">0</span><span class="p">),</span> <span class="n">axis</span><span class="o">=</span><span class="mi">0</span><span class="p">)</span>
<span class="k">def</span> <span class="nf">call</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">inputs</span><span class="p">):</span>
<span class="n">loss</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">triplet_loss</span><span class="p">(</span><span class="n">inputs</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">add_loss</span><span class="p">(</span><span class="n">loss</span><span class="p">)</span>
<span class="k">return</span> <span class="n">loss</span>
<span class="c"># Layer that computes the triplet loss from anchor, positive and negative embedding vectors</span>
<span class="n">triplet_loss_layer</span> <span class="o">=</span> <span class="n">TripletLossLayer</span><span class="p">(</span><span class="n">alpha</span><span class="o">=</span><span class="mf">0.2</span><span class="p">,</span> <span class="n">name</span><span class="o">=</span><span class="s">'triplet_loss_layer'</span><span class="p">)([</span><span class="n">emb_a</span><span class="p">,</span> <span class="n">emb_p</span><span class="p">,</span> <span class="n">emb_n</span><span class="p">])</span>
<span class="c"># Model that can be trained with anchor, positive negative images</span>
<span class="n">nn4_small2_train</span> <span class="o">=</span> <span class="n">Model</span><span class="p">([</span><span class="n">in_a</span><span class="p">,</span> <span class="n">in_p</span><span class="p">,</span> <span class="n">in_n</span><span class="p">],</span> <span class="n">triplet_loss_layer</span><span class="p">)</span>
</code></pre></div></div>
<p>During training, it is important to select triplets whose positive pairs <script type="math/tex">(x^a_i, x^p_i)</script> and negative pairs <script type="math/tex">(x^a_i, x^n_i)</script> are hard to discriminate i.e. their distance difference in embedding space should be less than margin <script type="math/tex">\alpha</script>, otherwise, the network is unable to learn a useful embedding. Therefore, each training iteration should select a new batch of triplets based on the embeddings learned in the previous iteration. Assuming that a generator returned from a <code class="highlighter-rouge">triplet_generator()</code> call can generate triplets under these constraints, the network can be trained with:</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">from</span> <span class="nn">data</span> <span class="kn">import</span> <span class="n">triplet_generator</span>
<span class="c"># triplet_generator() creates a generator that continuously returns </span>
<span class="c"># ([a_batch, p_batch, n_batch], None) tuples where a_batch, p_batch </span>
<span class="c"># and n_batch are batches of anchor, positive and negative RGB images </span>
<span class="c"># each having a shape of (batch_size, 96, 96, 3).</span>
<span class="n">generator</span> <span class="o">=</span> <span class="n">triplet_generator</span><span class="p">()</span>
<span class="n">nn4_small2_train</span><span class="o">.</span><span class="nb">compile</span><span class="p">(</span><span class="n">loss</span><span class="o">=</span><span class="bp">None</span><span class="p">,</span> <span class="n">optimizer</span><span class="o">=</span><span class="s">'adam'</span><span class="p">)</span>
<span class="n">nn4_small2_train</span><span class="o">.</span><span class="n">fit_generator</span><span class="p">(</span><span class="n">generator</span><span class="p">,</span> <span class="n">epochs</span><span class="o">=</span><span class="mi">10</span><span class="p">,</span> <span class="n">steps_per_epoch</span><span class="o">=</span><span class="mi">100</span><span class="p">)</span>
<span class="c"># Please note that the current implementation of the generator only generates </span>
<span class="c"># random image data. The main goal of this code snippet is to demonstrate </span>
<span class="c"># the general setup for model training. In the following, we will anyway </span>
<span class="c"># use a pre-trained model so we don't need a generator here that operates </span>
<span class="c"># on real training data. I'll maybe provide a fully functional generator</span>
<span class="c"># later.</span>
</code></pre></div></div>
<p>The above code snippet should merely demonstrate how to setup model training. But instead of actually training a model from scratch we will now use a pre-trained model as training from scratch is very expensive and requires huge datasets to achieve good generalization performance. For example, <a href="https://arxiv.org/abs/1503.03832">[1]</a> uses a dataset of 200M images consisting of about 8M identities.</p>
<p>The OpenFace project provides <a href="https://cmusatyalab.github.io/openface/models-and-accuracies/#pre-trained-models">pre-trained models</a> that were trained with the public face recognition datasets <a href="http://vintage.winklerbros.net/facescrub.html">FaceScrub</a> and <a href="http://arxiv.org/abs/1411.7923">CASIA-WebFace</a>. The Keras-OpenFace project converted the weights of the pre-trained nn4.small2.v1 model to <a href="https://github.com/iwantooxxoox/Keras-OpenFace/tree/master/weights">CSV files</a> which were then <a href="https://github.com/krasserm/face-recognition/blob/master/face-recognition-convert.ipynb">converted here</a> to a binary format that can be loaded by Keras with <code class="highlighter-rouge">load_weights</code>:</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="n">nn4_small2_pretrained</span> <span class="o">=</span> <span class="n">create_model</span><span class="p">()</span>
<span class="n">nn4_small2_pretrained</span><span class="o">.</span><span class="n">load_weights</span><span class="p">(</span><span class="s">'weights/nn4.small2.v1.h5'</span><span class="p">)</span>
</code></pre></div></div>
<h2 id="custom-dataset">Custom dataset</h2>
<p>To demonstrate face recognition on a custom dataset, a small subset of the <a href="http://vis-www.cs.umass.edu/lfw/">LFW</a> dataset is used. It consists of 100 face images of <a href="https://github.com/krasserm/face-recognition/tree/master/images">10 identities</a>. The metadata for each image (file and identity name) are loaded into memory for later processing.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">import</span> <span class="nn">numpy</span> <span class="k">as</span> <span class="n">np</span>
<span class="kn">import</span> <span class="nn">os.path</span>
<span class="k">class</span> <span class="nc">IdentityMetadata</span><span class="p">():</span>
<span class="k">def</span> <span class="nf">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">base</span><span class="p">,</span> <span class="n">name</span><span class="p">,</span> <span class="nb">file</span><span class="p">):</span>
<span class="c"># dataset base directory</span>
<span class="bp">self</span><span class="o">.</span><span class="n">base</span> <span class="o">=</span> <span class="n">base</span>
<span class="c"># identity name</span>
<span class="bp">self</span><span class="o">.</span><span class="n">name</span> <span class="o">=</span> <span class="n">name</span>
<span class="c"># image file name</span>
<span class="bp">self</span><span class="o">.</span><span class="nb">file</span> <span class="o">=</span> <span class="nb">file</span>
<span class="k">def</span> <span class="nf">__repr__</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
<span class="k">return</span> <span class="bp">self</span><span class="o">.</span><span class="n">image_path</span><span class="p">()</span>
<span class="k">def</span> <span class="nf">image_path</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
<span class="k">return</span> <span class="n">os</span><span class="o">.</span><span class="n">path</span><span class="o">.</span><span class="n">join</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">base</span><span class="p">,</span> <span class="bp">self</span><span class="o">.</span><span class="n">name</span><span class="p">,</span> <span class="bp">self</span><span class="o">.</span><span class="nb">file</span><span class="p">)</span>
<span class="k">def</span> <span class="nf">load_metadata</span><span class="p">(</span><span class="n">path</span><span class="p">):</span>
<span class="n">metadata</span> <span class="o">=</span> <span class="p">[]</span>
<span class="k">for</span> <span class="n">i</span> <span class="ow">in</span> <span class="n">os</span><span class="o">.</span><span class="n">listdir</span><span class="p">(</span><span class="n">path</span><span class="p">):</span>
<span class="k">for</span> <span class="n">f</span> <span class="ow">in</span> <span class="n">os</span><span class="o">.</span><span class="n">listdir</span><span class="p">(</span><span class="n">os</span><span class="o">.</span><span class="n">path</span><span class="o">.</span><span class="n">join</span><span class="p">(</span><span class="n">path</span><span class="p">,</span> <span class="n">i</span><span class="p">)):</span>
<span class="n">metadata</span><span class="o">.</span><span class="n">append</span><span class="p">(</span><span class="n">IdentityMetadata</span><span class="p">(</span><span class="n">path</span><span class="p">,</span> <span class="n">i</span><span class="p">,</span> <span class="n">f</span><span class="p">))</span>
<span class="k">return</span> <span class="n">np</span><span class="o">.</span><span class="n">array</span><span class="p">(</span><span class="n">metadata</span><span class="p">)</span>
<span class="n">metadata</span> <span class="o">=</span> <span class="n">load_metadata</span><span class="p">(</span><span class="s">'images'</span><span class="p">)</span>
</code></pre></div></div>
<h2 id="face-alignment">Face alignment</h2>
<p>The nn4.small2.v1 model was trained with aligned face images, therefore, the face images from the custom dataset must be aligned too. Here, we use <a href="http://dlib.net/">Dlib</a> for face detection and <a href="https://opencv.org/">OpenCV</a> for image transformation and cropping to produce aligned 96x96 RGB face images. By using the <a href="https://github.com/krasserm/face-recognition/blob/master/align.py">AlignDlib</a> utility from the OpenFace project this is straightforward:</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">import</span> <span class="nn">cv2</span>
<span class="kn">import</span> <span class="nn">matplotlib.pyplot</span> <span class="k">as</span> <span class="n">plt</span>
<span class="kn">import</span> <span class="nn">matplotlib.patches</span> <span class="k">as</span> <span class="n">patches</span>
<span class="kn">from</span> <span class="nn">align</span> <span class="kn">import</span> <span class="n">AlignDlib</span>
<span class="o">%</span><span class="n">matplotlib</span> <span class="n">inline</span>
<span class="k">def</span> <span class="nf">load_image</span><span class="p">(</span><span class="n">path</span><span class="p">):</span>
<span class="n">img</span> <span class="o">=</span> <span class="n">cv2</span><span class="o">.</span><span class="n">imread</span><span class="p">(</span><span class="n">path</span><span class="p">,</span> <span class="mi">1</span><span class="p">)</span>
<span class="c"># OpenCV loads images with color channels</span>
<span class="c"># in BGR order. So we need to reverse them</span>
<span class="k">return</span> <span class="n">img</span><span class="p">[</span><span class="o">...</span><span class="p">,::</span><span class="o">-</span><span class="mi">1</span><span class="p">]</span>
<span class="c"># Initialize the OpenFace face alignment utility</span>
<span class="n">alignment</span> <span class="o">=</span> <span class="n">AlignDlib</span><span class="p">(</span><span class="s">'models/landmarks.dat'</span><span class="p">)</span>
<span class="c"># Load an image of Jacques Chirac</span>
<span class="n">jc_orig</span> <span class="o">=</span> <span class="n">load_image</span><span class="p">(</span><span class="n">metadata</span><span class="p">[</span><span class="mi">2</span><span class="p">]</span><span class="o">.</span><span class="n">image_path</span><span class="p">())</span>
<span class="c"># Detect face and return bounding box</span>
<span class="n">bb</span> <span class="o">=</span> <span class="n">alignment</span><span class="o">.</span><span class="n">getLargestFaceBoundingBox</span><span class="p">(</span><span class="n">jc_orig</span><span class="p">)</span>
<span class="c"># Transform image using specified face landmark indices and crop image to 96x96</span>
<span class="n">jc_aligned</span> <span class="o">=</span> <span class="n">alignment</span><span class="o">.</span><span class="n">align</span><span class="p">(</span><span class="mi">96</span><span class="p">,</span> <span class="n">jc_orig</span><span class="p">,</span> <span class="n">bb</span><span class="p">,</span> <span class="n">landmarkIndices</span><span class="o">=</span><span class="n">AlignDlib</span><span class="o">.</span><span class="n">OUTER_EYES_AND_NOSE</span><span class="p">)</span>
<span class="c"># Show original image</span>
<span class="n">plt</span><span class="o">.</span><span class="n">subplot</span><span class="p">(</span><span class="mi">131</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">imshow</span><span class="p">(</span><span class="n">jc_orig</span><span class="p">)</span>
<span class="c"># Show original image with bounding box</span>
<span class="n">plt</span><span class="o">.</span><span class="n">subplot</span><span class="p">(</span><span class="mi">132</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">imshow</span><span class="p">(</span><span class="n">jc_orig</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">gca</span><span class="p">()</span><span class="o">.</span><span class="n">add_patch</span><span class="p">(</span><span class="n">patches</span><span class="o">.</span><span class="n">Rectangle</span><span class="p">((</span><span class="n">bb</span><span class="o">.</span><span class="n">left</span><span class="p">(),</span> <span class="n">bb</span><span class="o">.</span><span class="n">top</span><span class="p">()),</span> <span class="n">bb</span><span class="o">.</span><span class="n">width</span><span class="p">(),</span> <span class="n">bb</span><span class="o">.</span><span class="n">height</span><span class="p">(),</span> <span class="n">fill</span><span class="o">=</span><span class="bp">False</span><span class="p">,</span> <span class="n">color</span><span class="o">=</span><span class="s">'red'</span><span class="p">))</span>
<span class="c"># Show aligned image</span>
<span class="n">plt</span><span class="o">.</span><span class="n">subplot</span><span class="p">(</span><span class="mi">133</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">imshow</span><span class="p">(</span><span class="n">jc_aligned</span><span class="p">);</span>
</code></pre></div></div>
<p><img src="/img/2018-02-07/output_15_0.png" alt="png" /></p>
<p>As described in the OpenFace <a href="https://cmusatyalab.github.io/openface/models-and-accuracies/#pre-trained-models">pre-trained models</a> section, landmark indices <code class="highlighter-rouge">OUTER_EYES_AND_NOSE</code> are required for model nn4.small2.v1. Let’s implement face detection, transformation and cropping as <code class="highlighter-rouge">align_image</code> function for later reuse.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="k">def</span> <span class="nf">align_image</span><span class="p">(</span><span class="n">img</span><span class="p">):</span>
<span class="k">return</span> <span class="n">alignment</span><span class="o">.</span><span class="n">align</span><span class="p">(</span><span class="mi">96</span><span class="p">,</span> <span class="n">img</span><span class="p">,</span> <span class="n">alignment</span><span class="o">.</span><span class="n">getLargestFaceBoundingBox</span><span class="p">(</span><span class="n">img</span><span class="p">),</span>
<span class="n">landmarkIndices</span><span class="o">=</span><span class="n">AlignDlib</span><span class="o">.</span><span class="n">OUTER_EYES_AND_NOSE</span><span class="p">)</span>
</code></pre></div></div>
<h2 id="embedding-vectors">Embedding vectors</h2>
<p>Embedding vectors can now be calculated by feeding the aligned and scaled images into the pre-trained network.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="n">embedded</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">zeros</span><span class="p">((</span><span class="n">metadata</span><span class="o">.</span><span class="n">shape</span><span class="p">[</span><span class="mi">0</span><span class="p">],</span> <span class="mi">128</span><span class="p">))</span>
<span class="k">for</span> <span class="n">i</span><span class="p">,</span> <span class="n">m</span> <span class="ow">in</span> <span class="nb">enumerate</span><span class="p">(</span><span class="n">metadata</span><span class="p">):</span>
<span class="n">img</span> <span class="o">=</span> <span class="n">load_image</span><span class="p">(</span><span class="n">m</span><span class="o">.</span><span class="n">image_path</span><span class="p">())</span>
<span class="n">img</span> <span class="o">=</span> <span class="n">align_image</span><span class="p">(</span><span class="n">img</span><span class="p">)</span>
<span class="c"># scale RGB values to interval [0,1]</span>
<span class="n">img</span> <span class="o">=</span> <span class="p">(</span><span class="n">img</span> <span class="o">/</span> <span class="mf">255.</span><span class="p">)</span><span class="o">.</span><span class="n">astype</span><span class="p">(</span><span class="n">np</span><span class="o">.</span><span class="n">float32</span><span class="p">)</span>
<span class="c"># obtain embedding vector for image</span>
<span class="n">embedded</span><span class="p">[</span><span class="n">i</span><span class="p">]</span> <span class="o">=</span> <span class="n">nn4_small2_pretrained</span><span class="o">.</span><span class="n">predict</span><span class="p">(</span><span class="n">np</span><span class="o">.</span><span class="n">expand_dims</span><span class="p">(</span><span class="n">img</span><span class="p">,</span> <span class="n">axis</span><span class="o">=</span><span class="mi">0</span><span class="p">))[</span><span class="mi">0</span><span class="p">]</span>
</code></pre></div></div>
<p>Let’s verify on a single triplet example that the squared L2 distance between its anchor-positive pair is smaller than the distance between its anchor-negative pair.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="k">def</span> <span class="nf">distance</span><span class="p">(</span><span class="n">emb1</span><span class="p">,</span> <span class="n">emb2</span><span class="p">):</span>
<span class="k">return</span> <span class="n">np</span><span class="o">.</span><span class="nb">sum</span><span class="p">(</span><span class="n">np</span><span class="o">.</span><span class="n">square</span><span class="p">(</span><span class="n">emb1</span> <span class="o">-</span> <span class="n">emb2</span><span class="p">))</span>
<span class="k">def</span> <span class="nf">show_pair</span><span class="p">(</span><span class="n">idx1</span><span class="p">,</span> <span class="n">idx2</span><span class="p">):</span>
<span class="n">plt</span><span class="o">.</span><span class="n">figure</span><span class="p">(</span><span class="n">figsize</span><span class="o">=</span><span class="p">(</span><span class="mi">8</span><span class="p">,</span><span class="mi">3</span><span class="p">))</span>
<span class="n">plt</span><span class="o">.</span><span class="n">suptitle</span><span class="p">(</span><span class="n">f</span><span class="s">'Distance = {distance(embedded[idx1], embedded[idx2]):.2f}'</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">subplot</span><span class="p">(</span><span class="mi">121</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">imshow</span><span class="p">(</span><span class="n">load_image</span><span class="p">(</span><span class="n">metadata</span><span class="p">[</span><span class="n">idx1</span><span class="p">]</span><span class="o">.</span><span class="n">image_path</span><span class="p">()))</span>
<span class="n">plt</span><span class="o">.</span><span class="n">subplot</span><span class="p">(</span><span class="mi">122</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">imshow</span><span class="p">(</span><span class="n">load_image</span><span class="p">(</span><span class="n">metadata</span><span class="p">[</span><span class="n">idx2</span><span class="p">]</span><span class="o">.</span><span class="n">image_path</span><span class="p">()));</span>
<span class="n">show_pair</span><span class="p">(</span><span class="mi">2</span><span class="p">,</span> <span class="mi">3</span><span class="p">)</span>
<span class="n">show_pair</span><span class="p">(</span><span class="mi">2</span><span class="p">,</span> <span class="mi">12</span><span class="p">)</span>
</code></pre></div></div>
<p><img src="/img/2018-02-07/output_22_0.png" alt="png" /></p>
<p><img src="/img/2018-02-07/output_22_1.png" alt="png" /></p>
<p>As expected, the distance between the two images of Jacques Chirac is smaller than the distance between an image of Jacques Chirac and an image of Gerhard Schröder (0.30 &lt; 1.12). But we still do not know what distance threshold <script type="math/tex">\tau</script> is the best boundary for making a decision between <em>same identity</em> and <em>different identity</em>.</p>
<h2 id="distance-threshold">Distance threshold</h2>
<p>To find the optimal value for <script type="math/tex">\tau</script>, the face verification performance must be evaluated on a range of distance threshold values. At a given threshold, all possible embedding vector pairs are classified as either <em>same identity</em> or <em>different identity</em> and compared to the ground truth. Since we’re dealing with skewed classes (much more negative pairs than positive pairs), we use the <a href="https://en.wikipedia.org/wiki/F1_score">F1 score</a> as evaluation metric instead of <a href="http://scikit-learn.org/stable/modules/generated/sklearn.metrics.accuracy_score.html">accuracy</a>.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">from</span> <span class="nn">sklearn.metrics</span> <span class="kn">import</span> <span class="n">f1_score</span><span class="p">,</span> <span class="n">accuracy_score</span>
<span class="n">distances</span> <span class="o">=</span> <span class="p">[]</span> <span class="c"># squared L2 distance between pairs</span>
<span class="n">identical</span> <span class="o">=</span> <span class="p">[]</span> <span class="c"># 1 if same identity, 0 otherwise</span>
<span class="n">num</span> <span class="o">=</span> <span class="nb">len</span><span class="p">(</span><span class="n">metadata</span><span class="p">)</span>
<span class="k">for</span> <span class="n">i</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="n">num</span> <span class="o">-</span> <span class="mi">1</span><span class="p">):</span>
<span class="k">for</span> <span class="n">j</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="n">num</span><span class="p">):</span>
<span class="n">distances</span><span class="o">.</span><span class="n">append</span><span class="p">(</span><span class="n">distance</span><span class="p">(</span><span class="n">embedded</span><span class="p">[</span><span class="n">i</span><span class="p">],</span> <span class="n">embedded</span><span class="p">[</span><span class="n">j</span><span class="p">]))</span>
<span class="n">identical</span><span class="o">.</span><span class="n">append</span><span class="p">(</span><span class="mi">1</span> <span class="k">if</span> <span class="n">metadata</span><span class="p">[</span><span class="n">i</span><span class="p">]</span><span class="o">.</span><span class="n">name</span> <span class="o">==</span> <span class="n">metadata</span><span class="p">[</span><span class="n">j</span><span class="p">]</span><span class="o">.</span><span class="n">name</span> <span class="k">else</span> <span class="mi">0</span><span class="p">)</span>
<span class="n">distances</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">array</span><span class="p">(</span><span class="n">distances</span><span class="p">)</span>
<span class="n">identical</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">array</span><span class="p">(</span><span class="n">identical</span><span class="p">)</span>
<span class="n">thresholds</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">arange</span><span class="p">(</span><span class="mf">0.3</span><span class="p">,</span> <span class="mf">1.0</span><span class="p">,</span> <span class="mf">0.01</span><span class="p">)</span>
<span class="n">f1_scores</span> <span class="o">=</span> <span class="p">[</span><span class="n">f1_score</span><span class="p">(</span><span class="n">identical</span><span class="p">,</span> <span class="n">distances</span> <span class="o">&lt;</span> <span class="n">t</span><span class="p">)</span> <span class="k">for</span> <span class="n">t</span> <span class="ow">in</span> <span class="n">thresholds</span><span class="p">]</span>
<span class="n">acc_scores</span> <span class="o">=</span> <span class="p">[</span><span class="n">accuracy_score</span><span class="p">(</span><span class="n">identical</span><span class="p">,</span> <span class="n">distances</span> <span class="o">&lt;</span> <span class="n">t</span><span class="p">)</span> <span class="k">for</span> <span class="n">t</span> <span class="ow">in</span> <span class="n">thresholds</span><span class="p">]</span>
<span class="n">opt_idx</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">argmax</span><span class="p">(</span><span class="n">f1_scores</span><span class="p">)</span>
<span class="c"># Threshold at maximal F1 score</span>
<span class="n">opt_tau</span> <span class="o">=</span> <span class="n">thresholds</span><span class="p">[</span><span class="n">opt_idx</span><span class="p">]</span>
<span class="c"># Accuracy at maximal F1 score</span>
<span class="n">opt_acc</span> <span class="o">=</span> <span class="n">accuracy_score</span><span class="p">(</span><span class="n">identical</span><span class="p">,</span> <span class="n">distances</span> <span class="o">&lt;</span> <span class="n">opt_tau</span><span class="p">)</span>
<span class="c"># Plot F1 score and accuracy as function of distance threshold</span>
<span class="n">plt</span><span class="o">.</span><span class="n">plot</span><span class="p">(</span><span class="n">thresholds</span><span class="p">,</span> <span class="n">f1_scores</span><span class="p">,</span> <span class="n">label</span><span class="o">=</span><span class="s">'F1 score'</span><span class="p">);</span>
<span class="n">plt</span><span class="o">.</span><span class="n">plot</span><span class="p">(</span><span class="n">thresholds</span><span class="p">,</span> <span class="n">acc_scores</span><span class="p">,</span> <span class="n">label</span><span class="o">=</span><span class="s">'Accuracy'</span><span class="p">);</span>
<span class="n">plt</span><span class="o">.</span><span class="n">axvline</span><span class="p">(</span><span class="n">x</span><span class="o">=</span><span class="n">opt_tau</span><span class="p">,</span> <span class="n">linestyle</span><span class="o">=</span><span class="s">'--'</span><span class="p">,</span> <span class="n">lw</span><span class="o">=</span><span class="mi">1</span><span class="p">,</span> <span class="n">c</span><span class="o">=</span><span class="s">'lightgrey'</span><span class="p">,</span> <span class="n">label</span><span class="o">=</span><span class="s">'Threshold'</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">title</span><span class="p">(</span><span class="n">f</span><span class="s">'Accuracy at threshold {opt_tau:.2f} = {opt_acc:.3f}'</span><span class="p">);</span>
<span class="n">plt</span><span class="o">.</span><span class="n">xlabel</span><span class="p">(</span><span class="s">'Distance threshold'</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">legend</span><span class="p">();</span>
</code></pre></div></div>
<p><img src="/img/2018-02-07/output_26_0.png" alt="png" /></p>
<p>The face verification accuracy at <script type="math/tex">\tau</script> = 0.56 is 95.7%. This is not bad given a baseline of 89% for a classifier that always predicts <em>different identity</em> (there are 980 pos. pairs and 8821 neg. pairs) but since nn4.small2.v1 is a relatively small model it is still less than what can be achieved by state-of-the-art models (&gt; 99%).</p>
<p>The following two histograms show the distance distributions of positive and negative pairs and the location of the decision boundary. There is a clear separation of these distributions which explains the discriminative performance of the network. One can also spot some strong outliers in the positive pairs class but these are not further analyzed here.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="n">dist_pos</span> <span class="o">=</span> <span class="n">distances</span><span class="p">[</span><span class="n">identical</span> <span class="o">==</span> <span class="mi">1</span><span class="p">]</span>
<span class="n">dist_neg</span> <span class="o">=</span> <span class="n">distances</span><span class="p">[</span><span class="n">identical</span> <span class="o">==</span> <span class="mi">0</span><span class="p">]</span>
<span class="n">plt</span><span class="o">.</span><span class="n">figure</span><span class="p">(</span><span class="n">figsize</span><span class="o">=</span><span class="p">(</span><span class="mi">12</span><span class="p">,</span><span class="mi">4</span><span class="p">))</span>
<span class="n">plt</span><span class="o">.</span><span class="n">subplot</span><span class="p">(</span><span class="mi">121</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">hist</span><span class="p">(</span><span class="n">dist_pos</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">axvline</span><span class="p">(</span><span class="n">x</span><span class="o">=</span><span class="n">opt_tau</span><span class="p">,</span> <span class="n">linestyle</span><span class="o">=</span><span class="s">'--'</span><span class="p">,</span> <span class="n">lw</span><span class="o">=</span><span class="mi">1</span><span class="p">,</span> <span class="n">c</span><span class="o">=</span><span class="s">'lightgrey'</span><span class="p">,</span> <span class="n">label</span><span class="o">=</span><span class="s">'Threshold'</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">title</span><span class="p">(</span><span class="s">'Distances (pos. pairs)'</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">legend</span><span class="p">();</span>
<span class="n">plt</span><span class="o">.</span><span class="n">subplot</span><span class="p">(</span><span class="mi">122</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">hist</span><span class="p">(</span><span class="n">dist_neg</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">axvline</span><span class="p">(</span><span class="n">x</span><span class="o">=</span><span class="n">opt_tau</span><span class="p">,</span> <span class="n">linestyle</span><span class="o">=</span><span class="s">'--'</span><span class="p">,</span> <span class="n">lw</span><span class="o">=</span><span class="mi">1</span><span class="p">,</span> <span class="n">c</span><span class="o">=</span><span class="s">'lightgrey'</span><span class="p">,</span> <span class="n">label</span><span class="o">=</span><span class="s">'Threshold'</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">title</span><span class="p">(</span><span class="s">'Distances (neg. pairs)'</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">legend</span><span class="p">();</span>
</code></pre></div></div>
<p><img src="/img/2018-02-07/output_28_0.png" alt="png" /></p>
<h2 id="face-recognition">Face recognition</h2>
<p>Given an estimate of the distance threshold <script type="math/tex">\tau</script>, face recognition is now as simple as calculating the distances between an input embedding vector and all embedding vectors in a database. The input is assigned the label (i.e. identity) of the database entry with the smallest distance if it is less than <script type="math/tex">\tau</script> or label <em>unknown</em> otherwise. This procedure can also scale to large databases as it can be easily parallelized. It also supports one-shot learning, as adding only a single entry of a new identity might be sufficient to recognize new examples of that identity.</p>
<p>A more robust approach is to label the input using the top <script type="math/tex">k</script> scoring entries in the database which is essentially <a href="https://en.wikipedia.org/wiki/K-nearest_neighbors_algorithm">KNN classification</a> with a Euclidean distance metric. Alternatively, a linear <a href="https://en.wikipedia.org/wiki/Support_vector_machine">support vector machine</a> (SVM) can be trained with the database entries and used to classify i.e. identify new inputs. For training these classifiers we use 50% of the dataset, for evaluation the other 50%.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">from</span> <span class="nn">sklearn.preprocessing</span> <span class="kn">import</span> <span class="n">LabelEncoder</span>
<span class="kn">from</span> <span class="nn">sklearn.neighbors</span> <span class="kn">import</span> <span class="n">KNeighborsClassifier</span>
<span class="kn">from</span> <span class="nn">sklearn.svm</span> <span class="kn">import</span> <span class="n">LinearSVC</span>
<span class="n">targets</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">array</span><span class="p">([</span><span class="n">m</span><span class="o">.</span><span class="n">name</span> <span class="k">for</span> <span class="n">m</span> <span class="ow">in</span> <span class="n">metadata</span><span class="p">])</span>
<span class="n">encoder</span> <span class="o">=</span> <span class="n">LabelEncoder</span><span class="p">()</span>
<span class="n">encoder</span><span class="o">.</span><span class="n">fit</span><span class="p">(</span><span class="n">targets</span><span class="p">)</span>
<span class="c"># Numerical encoding of identities</span>
<span class="n">y</span> <span class="o">=</span> <span class="n">encoder</span><span class="o">.</span><span class="n">transform</span><span class="p">(</span><span class="n">targets</span><span class="p">)</span>
<span class="n">train_idx</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">arange</span><span class="p">(</span><span class="n">metadata</span><span class="o">.</span><span class="n">shape</span><span class="p">[</span><span class="mi">0</span><span class="p">])</span> <span class="o">%</span> <span class="mi">2</span> <span class="o">!=</span> <span class="mi">0</span>
<span class="n">test_idx</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">arange</span><span class="p">(</span><span class="n">metadata</span><span class="o">.</span><span class="n">shape</span><span class="p">[</span><span class="mi">0</span><span class="p">])</span> <span class="o">%</span> <span class="mi">2</span> <span class="o">==</span> <span class="mi">0</span>
<span class="c"># 50 train examples of 10 identities (5 examples each)</span>
<span class="n">X_train</span> <span class="o">=</span> <span class="n">embedded</span><span class="p">[</span><span class="n">train_idx</span><span class="p">]</span>
<span class="c"># 50 test examples of 10 identities (5 examples each)</span>
<span class="n">X_test</span> <span class="o">=</span> <span class="n">embedded</span><span class="p">[</span><span class="n">test_idx</span><span class="p">]</span>
<span class="n">y_train</span> <span class="o">=</span> <span class="n">y</span><span class="p">[</span><span class="n">train_idx</span><span class="p">]</span>
<span class="n">y_test</span> <span class="o">=</span> <span class="n">y</span><span class="p">[</span><span class="n">test_idx</span><span class="p">]</span>
<span class="n">knn</span> <span class="o">=</span> <span class="n">KNeighborsClassifier</span><span class="p">(</span><span class="n">n_neighbors</span><span class="o">=</span><span class="mi">1</span><span class="p">,</span> <span class="n">metric</span><span class="o">=</span><span class="s">'euclidean'</span><span class="p">)</span>
<span class="n">svc</span> <span class="o">=</span> <span class="n">LinearSVC</span><span class="p">()</span>
<span class="n">knn</span><span class="o">.</span><span class="n">fit</span><span class="p">(</span><span class="n">X_train</span><span class="p">,</span> <span class="n">y_train</span><span class="p">)</span>
<span class="n">svc</span><span class="o">.</span><span class="n">fit</span><span class="p">(</span><span class="n">X_train</span><span class="p">,</span> <span class="n">y_train</span><span class="p">)</span>
<span class="n">acc_knn</span> <span class="o">=</span> <span class="n">accuracy_score</span><span class="p">(</span><span class="n">y_test</span><span class="p">,</span> <span class="n">knn</span><span class="o">.</span><span class="n">predict</span><span class="p">(</span><span class="n">X_test</span><span class="p">))</span>
<span class="n">acc_svc</span> <span class="o">=</span> <span class="n">accuracy_score</span><span class="p">(</span><span class="n">y_test</span><span class="p">,</span> <span class="n">svc</span><span class="o">.</span><span class="n">predict</span><span class="p">(</span><span class="n">X_test</span><span class="p">))</span>
<span class="k">print</span><span class="p">(</span><span class="n">f</span><span class="s">'KNN accuracy = {acc_knn}, SVM accuracy = {acc_svc}'</span><span class="p">)</span>
</code></pre></div></div>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>KNN accuracy = 0.96, SVM accuracy = 0.98
</code></pre></div></div>
<p>The KNN classifier achieves an accuracy of 96% on the test set, the SVM classifier 98%. Let’s use the SVM classifier to illustrate face recognition on a single example.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">import</span> <span class="nn">warnings</span>
<span class="c"># Suppress LabelEncoder warning</span>
<span class="n">warnings</span><span class="o">.</span><span class="n">filterwarnings</span><span class="p">(</span><span class="s">'ignore'</span><span class="p">)</span>
<span class="n">example_idx</span> <span class="o">=</span> <span class="mi">29</span>
<span class="n">example_image</span> <span class="o">=</span> <span class="n">load_image</span><span class="p">(</span><span class="n">metadata</span><span class="p">[</span><span class="n">test_idx</span><span class="p">][</span><span class="n">example_idx</span><span class="p">]</span><span class="o">.</span><span class="n">image_path</span><span class="p">())</span>
<span class="n">example_prediction</span> <span class="o">=</span> <span class="n">svc</span><span class="o">.</span><span class="n">predict</span><span class="p">([</span><span class="n">embedded</span><span class="p">[</span><span class="n">test_idx</span><span class="p">][</span><span class="n">example_idx</span><span class="p">]])</span>
<span class="n">example_identity</span> <span class="o">=</span> <span class="n">encoder</span><span class="o">.</span><span class="n">inverse_transform</span><span class="p">(</span><span class="n">example_prediction</span><span class="p">)[</span><span class="mi">0</span><span class="p">]</span>
<span class="n">plt</span><span class="o">.</span><span class="n">imshow</span><span class="p">(</span><span class="n">example_image</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">title</span><span class="p">(</span><span class="n">f</span><span class="s">'Recognized as {example_identity}'</span><span class="p">);</span>
</code></pre></div></div>
<p><img src="/img/2018-02-07/output_33_0.png" alt="png" /></p>
<p>Seems reasonable :-) Classification results should actually be checked whether (a subset of) the database entries of the predicted identity have a distance less than <script type="math/tex">\tau</script>, otherwise one should assign an <em>unknown</em> label. This step is skipped here but can be easily added.</p>
<h2 id="dataset-visualization">Dataset visualization</h2>
<p>To embed the dataset into 2D space for displaying identity clusters, <a href="https://en.wikipedia.org/wiki/T-distributed_stochastic_neighbor_embedding">t-distributed Stochastic Neighbor Embedding</a> (t-SNE) is applied to the 128-dimensional embedding vectors. Except from a few outliers, identity clusters are well separated.</p>
<div class="language-python highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="kn">from</span> <span class="nn">sklearn.manifold</span> <span class="kn">import</span> <span class="n">TSNE</span>
<span class="n">X_embedded</span> <span class="o">=</span> <span class="n">TSNE</span><span class="p">(</span><span class="n">n_components</span><span class="o">=</span><span class="mi">2</span><span class="p">)</span><span class="o">.</span><span class="n">fit_transform</span><span class="p">(</span><span class="n">embedded</span><span class="p">)</span>
<span class="k">for</span> <span class="n">i</span><span class="p">,</span> <span class="n">t</span> <span class="ow">in</span> <span class="nb">enumerate</span><span class="p">(</span><span class="nb">set</span><span class="p">(</span><span class="n">targets</span><span class="p">)):</span>
<span class="n">idx</span> <span class="o">=</span> <span class="n">targets</span> <span class="o">==</span> <span class="n">t</span>
<span class="n">plt</span><span class="o">.</span><span class="n">scatter</span><span class="p">(</span><span class="n">X_embedded</span><span class="p">[</span><span class="n">idx</span><span class="p">,</span> <span class="mi">0</span><span class="p">],</span> <span class="n">X_embedded</span><span class="p">[</span><span class="n">idx</span><span class="p">,</span> <span class="mi">1</span><span class="p">],</span> <span class="n">label</span><span class="o">=</span><span class="n">t</span><span class="p">)</span>
<span class="n">plt</span><span class="o">.</span><span class="n">legend</span><span class="p">(</span><span class="n">bbox_to_anchor</span><span class="o">=</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="mi">1</span><span class="p">));</span>
</code></pre></div></div>
<p><img src="/img/2018-02-07/output_37_0.png" alt="png" /></p>
<h2 id="references">References</h2>
<ul>
<li>[1] <a href="https://arxiv.org/abs/1503.03832">FaceNet: A Unified Embedding for Face Recognition and Clustering</a></li>
<li>[2] <a href="https://arxiv.org/abs/1409.4842">Going Deeper with Convolutions</a></li>
</ul>
Wed, 07 Feb 2018 00:00:00 +0000http://krasserm.github.io/2018/02/07/deep-face-recognition/
http://krasserm.github.io/2018/02/07/deep-face-recognition/Resources for getting started with ML and DL<p>The following is a collection of resources that I found useful when I started to learn machine learning and deep learning.
In this post I’m using the term <em>machine learning</em> to refer to classical machine learning and the term <em>deep learning</em>
to refer to machine learning with deep neural networks. There are numerous other good resources out there which are not
mentioned here. This doesn’t mean I consider them as inferior, it’s just that I haven’t used them and therefore can’t
comment on.</p>
<h2 id="first-steps">First steps</h2>
<p>If you are completely new to machine learning I recommend starting with the outstanding Stanford
<a href="https://www.coursera.org/learn/machine-learning">Machine Learning course</a> by Andrew Ng. It is easy to follow and covers
topics that every machine learning engineer really should know. The course uses Octave (an open source alternative to MATLAB)
for programming. Algorithms are implemented from scratch in order to get a better understanding how they work. The course
is also a good preparation for the <a href="https://www.coursera.org/specializations/deep-learning">Deep Learning specialization</a>
at Coursera.</p>
<p>After having taken the course I felt the need to learn Python and re-implement the exercises with <a href="http://scikit-learn.org">scikit-learn</a>.
Scikit-learn is a Python machine learning library that provides optimized and easy-to-use implementations for all algorithms
presented in the course. I published the results as <a href="https://github.com/krasserm/machine-learning-notebooks">machine-learning-notebooks</a>
project on GitHub.</p>
<p>If you are new to Python, the <a href="https://docs.python.org/3/tutorial/">Python tutorial</a> is a great resource to start with.
I also recommend to work at least through the <a href="https://docs.scipy.org/doc/numpy-dev/user/quickstart.html">NumPy tutorial</a>,
<a href="https://docs.scipy.org/doc/scipy/reference/tutorial/index.html">SciPy tutorial</a>,
<a href="http://pandas.pydata.org/pandas-docs/stable/10min.html">Pandas tutorial</a> and
<a href="http://matplotlib.org/users/pyplot_tutorial.html">Pyplot tutorial</a> before starting with the
<a href="http://scikit-learn.org/stable/tutorial/index.html">scikit-learn tutorials</a>. After having worked through these tutorials
you should be prepared for implementing the algorithms presented in the course with scikit-learn.</p>
<h2 id="further-courses">Further courses</h2>
<ul>
<li>
<p><a href="https://www.coursera.org/specializations/deep-learning">Deep Learning specialization</a>. This specialization consists of
five courses, tought by Andrew Ng, covering deep neural network basics, regularization and optimization and models for
computer vision and sequences (text, speech, …). If you enjoyed the quality and accessibility of Andrew’s
<a href="https://www.coursera.org/learn/machine-learning">Machine Learning course</a> you will probably like this course too. It
provides you with the skills needed to follow more advanced literature in that field, including research papers.</p>
<p>The initial programming exercises for the basics are in plain Python/numpy to get a better understanding how forward
and backward propagation work. Models for computer vision are implemented with <a href="https://www.tensorflow.org/">Tensorflow</a>
and <a href="https://keras.io/">Keras</a>. Many examples cover recent research literature from 2014 or newer (ResNet, GoogLeNet,
FaceNet, … and many more). The last course on sequence models wasn’t available yet at the time of writing this post.</p>
</li>
</ul>
<p>A good understanding of statistical inference basics is important to get more out of the machine learning, deep learning
and statistics literature listed further below. If you need a refresher on statistical inference basics then the following
courses might be helpful:</p>
<ul>
<li>
<p><a href="https://www.coursera.org/learn/inferential-statistics-intro">Inferential statistics</a>. This course covers the basics of
inference for numerical and categorical data, hypothesis testing and statistical tests such as ANOVA and Chi-squared.
It follows the <a href="https://en.wikipedia.org/wiki/Frequentist_inference">frequentist approach</a> to statistical inference and
is part of the <a href="https://www.coursera.org/specializations/statistics">Statistics with R specialization</a>. The course content
(except R basics) is also covered by the freely available book <a href="https://www.openintro.org/stat/textbook.php">OpenIntro Statistics</a>.</p>
</li>
<li>
<p><a href="https://www.coursera.org/learn/bayesian-statistics">Bayesian statistics</a>. Many advanced machine learning and deep
learning techniques are based on <a href="https://en.wikipedia.org/wiki/Bayesian_inference">Bayesian inference</a>. The course
teaches the basics (Bayes’ rule, conjugate models, Bayesian inference on discrete and continuous data, …) and compares
them to the frequentist approach. Other basics such as Markov Chain Monte Carlo (MCMC) and hierarchical models are not
covered though. A good companion to this course is the book
<a href="https://www.amazon.com/Doing-Bayesian-Data-Analysis-Second/dp/0124058884">Doing Bayesian data analysis</a> (see below).
Before taking this course, familiarity with the frequentist approach is helpful.</p>
</li>
</ul>
<h2 id="books">Books</h2>
<ul>
<li>
<p><a href="https://mitpress.mit.edu/books/machine-learning-0">Machine learning - a probabilistic perspective</a>. A comprehensive
book on classical machine learning techniques. Its focus is rather theoretical and the descriptions are math-heavy. All
concepts are explained in an excellent way and therefore rather easy to follow even for machine learning beginners, given
basic familiarity with multivariate calculus, probability and linear algebra. The book covers both the frequentist and
Bayesian approach to inferring parameters of statistical models. Code examples are in MATLAB but there is also a
<a href="https://github.com/probml/pyprobml">Python port</a> available.</p>
</li>
<li>
<p><a href="http://www.deeplearningbook.org/">Deep learning</a>. A comprehensive book on deep learning techniques. Part 1 covers
machine learning basics. Part 2 covers deep neural network basics, convolutional neural networks (CNNs) and recurrent
neural networks (RNNs). The content is comparable to that of the
<a href="https://www.coursera.org/specializations/deep-learning">Deep Learning specialization</a> but is presented in a more academic
way. Part 3 covers more advanced topics such as auto-encoders, representation learning and deep generative models. This
is not a book for a practitioners but one of the best deep learning overview books I’ve seen.</p>
</li>
<li>
<p><a href="http://shop.oreilly.com/product/0636920052289.do">Hands-on machine learning with scikit-learn and Tensorflow</a>. If
you’ve already taken a first machine learning and deep learning course, this book is for you. It is packed with useful
code examples and guidelines for real-world machine learning projects. Part 1 focuses on the implementation of classical
machine learning models with scikit-learn. Part 2 focuses on deep learning with Tensorflow. In addition to CNNs and RNNs
this part also has chapters on auto-encoders and reinforcement learning. Both, theory and code examples are presented
in a clear and concise way.</p>
</li>
<li>
<p><a href="https://www.manning.com/books/deep-learning-with-python">Deep learning with Python</a>. Another excellent deep learning
book for practitioners with code examples using Keras. Keras is a deep learning framework with a higher-level API than
Tensorflow that aims to enable rapid prototyping. In addition to a detailed coverage of CNNs and RNNs this book also has
chapters on advanced deep learning best practices and generative deep learning. It is a good complement to part 2 of the
previous books (from a tools perspective). If you are not sure which one is better to start with, I recommend this one as
first steps are easier with Keras than with Tensorflow in my opinion.</p>
</li>
<li>
<p><a href="http://www.springer.com/book/9781461471370">Introduction to statistical learning</a>. If
<a href="https://mitpress.mit.edu/books/machine-learning-0">Machine learning - a probabilistic perspective</a> is too math-heavy for
you, this book is a good alternative. It covers statistical machine learning basics with a minimum of maths and approaches
it from a frequentist inference perspective. It is also an excellent introduction to R. If you want to go deeper after
having read this book, both in terms of math and number of approaches, I recommend
<a href="http://www.springer.com/book/9780387848570">The elements of statistical learning</a>. Both books are also freely available
as PDF (<a href="http://www-bcf.usc.edu/~gareth/ISL/ISLR%20Seventh%20Printing.pdf">ISL</a>,
<a href="https://web.stanford.edu/~hastie/ElemStatLearn/printings/ESLII_print12.pdf">ESL</a>).</p>
</li>
<li>
<p><a href="https://www.amazon.com/Doing-Bayesian-Data-Analysis-Second/dp/0124058884">Doing Bayesian data analysis</a>. An excellent
introduction to Bayesian statistics that prepares you well for reading more advanced literature in that field. It covers
the Bayesian analogues to traditional statistical tests (t, ANOVA, Chi-squared, …) and to multiple linear and logistic
regression among many others. It requires only a basic knowledge of calculus. For me, the book was a helpful companion to
the books <a href="https://mitpress.mit.edu/books/machine-learning-0">Machine learning - a probabilistic perspective</a> and
<a href="http://www.deeplearningbook.org/">Deep learning</a>. Code examples are written in R using packages
<a href="http://mcmc-jags.sourceforge.net/">JAGS</a> and <a href="http://mc-stan.org/users/interfaces/rstan">Stan</a> for MCMC sampling. There’s
also a <a href="https://github.com/JWarmenhoven/DBDA-python">Python port</a> available using <a href="http://docs.pymc.io/">PyMC3</a>.</p>
</li>
<li>
<p><a href="http://shop.oreilly.com/product/0636920033400.do">Data Science from Scratch</a>. This book is about data science in its
most distilled form. Don’t expect too much depth here but a great overview of data science topics such as probability and
statistics, data preparation and machine learning basics. The book focuses on understanding fundamental data science tools
by implementing them in plain Python from scratch. Well-known statistical and machine learning libraries are not used here
but each chapter contains references to libraries you should actually use for your own projects and links for further reading.</p>
</li>
</ul>
<p>There is also a large number of useful blogs and survey papers about machine learning which I’ll leave for a separate
post. I nevertheless hope you find this a useful guide for getting started with machine learning.</p>
Wed, 03 Jan 2018 00:00:00 +0000http://krasserm.github.io/2018/01/03/machine-learning-resources/
http://krasserm.github.io/2018/01/03/machine-learning-resources/A service framework for operation-based CRDTs<p>This article introduces a framework for developing operation-based CRDT services. It is part of the <a href="https://github.com/RBMHTechnology/eventuate">Eventuate</a> project and supports the integration of operation-based CRDTs into Eventuate’s reliable causal broadcast infrastructure. It exposes CRDT instances through an asynchronous service API, manages their durability and recovery and allows for CRDT replication up to global scale.</p>
<p>After a brief introduction to operation-based CRDTs, their relation to Eventuate’s <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#event-sourcing">event sourcing</a> and <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#event-collaboration">event collaboration</a> features is discussed to get a better understanding of the inner workings of the framework. An example then demonstrates how to use the framework to implement a concrete CRDT service and how to run multiple replicas of that service. The rest of the article covers production deployment considerations and gives an overview of planned features.</p>
<h2 id="operation-based-crdts">Operation-based CRDTs</h2>
<p>Conflict-free replicated data types (CRDTs) are replicated data types that eventually converge to the same state under concurrent updates. A CRDT instance can be updated without requiring coordination with its replicas. This makes CRDTs highly available for writes. CRDTs can be classified into state-based CRDTs (CvRDTs or convergent replicated data types) and operation-based CRDTs (CmRDTs or commutative replicated data types). State-based CRDTs are designed to disseminate state among replicas whereas operation-based CRDTs are designed to disseminate operations.</p>
<ul>
<li>
<p>CmRDT replicas are guaranteed to converge if operations are disseminated through a reliable causal broadcast (RCB) middleware and if they are designed to be commutative for concurrent operations.</p>
</li>
<li>
<p>CvRDTs don’t require special guarantees from the underlying messaging middleware but require increasing network bandwidth for state dissemination with increasing state size. They converge if their state <em>merge</em> function is defined as a join: a least upper bound on a <a href="https://en.wikipedia.org/wiki/Join-semilattice">join-semilattice</a>. CvRDTs are not further discussed in this article.</p>
</li>
</ul>
<p>The execution of a CmRDT operation is done in two phases, <em>prepare</em> and <em>effect</em> <a href="http://haslab.uminho.pt/sites/default/files/ashoker/files/opbaseddais14.pdf">[2]</a> (also called <em>atSource</em> and <em>downstream</em> in <a href="http://hal.upmc.fr/file/index/docid/555588/filename/techreport.pdf">[1]</a>): <em>prepare</em> is executed on the local replica. It looks at the operation and (optionally) the current state and produces a message, representing the operation, that is then disseminated to all replicas. <em>effect</em> applies the disseminated operation at all replicas.</p>
<h2 id="relation-to-event-sourcing">Relation to event sourcing</h2>
<p>The two CmRDT update phases, <em>prepare</em> and <em>effect</em>, are closely related to the update phases of event-sourced entities, <em>command handling</em> and <em>event handling</em>, respectively:</p>
<ul>
<li>
<p>During <em>command handling</em>, an incoming command is (optionally) validated against current state of the entity and, if validation succeeds, an event representing the effect of the command is written to the event log. This corresponds to producing an operation representation during the CmRDT <em>prepare</em> phase.</p>
</li>
<li>
<p>During <em>event handling</em>, the written event is consumed from the event log and used to update the current state of the entity. This corresponds to applying the produced operation representation to CmRDT state locally during the effect <em>phase</em>.</p>
</li>
</ul>
<p>Event sourcing toolkits usually distinguish these two phases and provide event-sourced entity abstractions that let application define custom command and event handlers. For example, <a href="http://doc.akka.io/docs/akka/2.4/scala/persistence.html">Akka Persistence</a> provides <a href="http://doc.akka.io/api/akka/2.4/#akka.persistence.PersistentActor">PersistentActor</a>, Eventuate provides <a href="http://rbmhtechnology.github.io/eventuate/latest/api/index.html#com.rbmhtechnology.eventuate.EventsourcedActor">EventsourcedActor</a>.</p>
<p>These are quite similar from a conceptual and API perspective but have one important difference: A <code class="highlighter-rouge">PersistentActor</code> can only consume self-emitted events whereas an <code class="highlighter-rouge">EventsourcedActor</code> can also consume events emitted by other event-sourced actors. And exactly this is needed to implement CmRDTs with event-sourced entities. They must be able to <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#event-collaboration">collaborate</a> so that multiple replicas of the same entity can be deployed.</p>
<p>CmRDTs in Eventuate are plain Scala objects that are <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#operation-based-crdts">internally managed inside event-sourced CRDT actors</a>. CmRDTs encapsulate state and implement <em>prepare</em> and <em>effect</em> logic. Their concrete implementation follows the specifications in <a href="http://hal.upmc.fr/file/index/docid/555588/filename/techreport.pdf">[1]</a>. CRDT actors handle CmRDT update commands by persisting update operations to the event log in the <em>prepare</em> phase and delegate the execution of these operations to the CmRDT in the <em>effect</em> phase. An operation emitted by one event-sourced actor is not only consumed by that actor itself but also by all other event-sourced actors with the same <code class="highlighter-rouge">aggregateId</code>. This ensures that all replicas of a given CmRDT instance receive the update operation during the <em>effect</em> phase.</p>
<h2 id="reliable-causal-broadcast">Reliable causal broadcast</h2>
<p>Most CmRDTs specified in <a href="http://hal.upmc.fr/file/index/docid/555588/filename/techreport.pdf">[1]</a> require causal delivery order for update operations. Causal delivery is trivial to achieve with an event log that provides total order. But the availability of such an event log is limited because all updates to it must be coordinated if the event log itself is replicated (like a topic partition in a Kafka cluster, for example) or because it is not replicated at all. Consequently, the availability of CmRDTs that share a totally ordered event log is constrained by the availability of the underlying event log, which is not what we want.</p>
<p>What we want is a distribution of CmRDT replicas across <em>locations</em> (or availability zones) where each location has its own local event log that remains available for writes even if partitioned from other locations. Events written at one location are asynchronously and reliably replicated to other locations. The strongest global ordering that can be achieved in such a <em>replication network</em> of local event logs is <em>causal ordering</em> which is sufficient for CmRDTs to work as specified.</p>
<p>A replication network of local event logs in Eventuate is called a <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#event-logs">replicated event log</a>. Causality in a replicated event log is tracked with <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#vector-clocks">vector clocks</a> as <em>potential causality</em> (a partial order). The total order of events in a local event log at each location is consistent with potential causality. Events produced within a replication network can therefore be consumed in correct causal order at any location of that network. This property makes replicated event logs a reliable causal broadcast (RCB) mechanism as required by CmRDTs.</p>
<p>Inter-location event replication is Eventuate-specific and ensures that local storage order is consistent with potential causality. The pluggable <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#storage-backends">storage backend</a> of a local event log may additionally provide redundant storage of events if stronger durability is needed. Event replication within a storage backend is completely independent from inter-location event replication. Inter-location event replication also works between locations with different storage backends.</p>
<h2 id="crdt-service-framework">CRDT service framework</h2>
<h3 id="crdt-service-definition">CRDT service definition</h3>
<p>The previous sections outlined how CmRDTs can be integrated into Eventuate’s event sourcing and event collaboration infrastructure. In order to free CmRDT developers and users from working with actor APIs directly, Eventuate provides a CmRDT service development framework that hides these low-level details. This framework is also the implementation basis for all CmRDTs that are part of Eventuate. Usage of the framework is demonstrated in the following by implementing an <em>MV-Register</em> CRDT service (full source code <a href="https://github.com/RBMHTechnology/eventuate/blob/master/eventuate-crdt/src/main/scala/com/rbmhtechnology/eventuate/crdt/MVRegister.scala">here</a>).</p>
<p>An <em>MV-Register</em> or <em>Multi-Value Register</em> CRDT is a memory cell that supports <em>assign</em> to update the cell value and <em>value</em> to query it. In case of concurrent updates, multiple values are retained (in contrast to a <em>Last-Writer-Wins Register</em>, for example, where one of the values takes precedence). In case of multiple values, applications can later reduce them to a single value.</p>
<div class="language-scala highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.VectorTime</span>
<span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.Versioned</span>
<span class="k">case</span> <span class="k">class</span> <span class="nc">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">](</span><span class="n">versioned</span><span class="k">:</span> <span class="kt">Set</span><span class="o">[</span><span class="kt">Versioned</span><span class="o">[</span><span class="kt">A</span><span class="o">]]</span> <span class="k">=</span> <span class="nc">Set</span><span class="o">.</span><span class="n">empty</span><span class="o">[</span><span class="kt">Versioned</span><span class="o">[</span><span class="kt">A</span><span class="o">]])</span> <span class="o">{</span>
<span class="k">def</span> <span class="n">value</span><span class="k">:</span> <span class="kt">Set</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span> <span class="k">=</span> <span class="n">versioned</span><span class="o">.</span><span class="n">map</span><span class="o">(</span><span class="k">_</span><span class="o">.</span><span class="n">value</span><span class="o">)</span>
<span class="k">def</span> <span class="n">assign</span><span class="o">(</span><span class="n">v</span><span class="k">:</span> <span class="kt">A</span><span class="o">,</span> <span class="n">vectorTimestamp</span><span class="k">:</span> <span class="kt">VectorTime</span><span class="o">)</span><span class="k">:</span> <span class="kt">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span> <span class="k">=</span> <span class="o">{</span>
<span class="c1">// &lt;-&gt; operator returns true for concurrent vector timestamps
</span>
<span class="k">val</span> <span class="n">concurrent</span> <span class="k">=</span> <span class="n">versioned</span><span class="o">.</span><span class="n">filter</span><span class="o">(</span><span class="k">_</span><span class="o">.</span><span class="n">vectorTimestamp</span> <span class="o">&lt;-&gt;</span> <span class="n">vectorTimestamp</span><span class="o">)</span>
<span class="n">copy</span><span class="o">(</span><span class="n">concurrent</span> <span class="o">+</span> <span class="nc">Versioned</span><span class="o">(</span><span class="n">v</span><span class="o">,</span> <span class="n">vectorTimestamp</span><span class="o">))</span>
<span class="o">}</span>
<span class="o">}</span>
</code></pre></div></div>
<p>The <code class="highlighter-rouge">MVRegister[A]</code> class internally stores assigned values of type <code class="highlighter-rouge">A</code> together with the <code class="highlighter-rouge">vectorTimestamp</code> of the corresponding operation as <code class="highlighter-rouge">Versioned[A](value: A, vectorTimestamp: VectorTime)</code>. Here, <code class="highlighter-rouge">Set[Versioned[A]]</code> is used because multiple concurrent assignments are allowed. Vector timestamps are generated by Eventuate for each persisted event (i.e. operation) and can be used by applications to determine whether any two events are causally related or concurrent. The <code class="highlighter-rouge">assign</code> method retains all assignments that are concurrent to the new assignment. All other existing assignments can be assumed to causally precede the new assignment because of causal delivery order. The new assignment is then added to retained concurrent assignments. The current value of an <code class="highlighter-rouge">MVRegister[A]</code> can be obtained with <code class="highlighter-rouge">value</code>.</p>
<p>To provide a service API for <code class="highlighter-rouge">MVRegister[A]</code>, we define a <code class="highlighter-rouge">MVRegisterService[A]</code> class that extends <code class="highlighter-rouge">CRDTService[MVRegister[A], Set[A]]</code>. <a href="http://rbmhtechnology.github.io/eventuate/latest/api/index.html#com.rbmhtechnology.eventuate.crdt.CRDTService">CRDTService</a> is an abstract service that manages CmRDT instances inside event-sourced actors and provides an asynchronous API for reading and updating those instances. The service-internal actors interact with the local event log for reading and writing CmRDT operations. The only concrete service method we need to implement is the asynchronous <code class="highlighter-rouge">assign</code> method that defines which operation to use for the <code class="highlighter-rouge">MVRegister[A]</code> update.</p>
<div class="language-scala highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="k">import</span> <span class="nn">akka.actor.ActorRef</span>
<span class="k">import</span> <span class="nn">akka.actor.ActorSystem</span>
<span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.crdt.CRDTService</span>
<span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.crdt.CRDTServiceOps</span>
<span class="k">import</span> <span class="nn">scala.concurrent.Future</span>
<span class="k">class</span> <span class="nc">MVRegisterService</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span>
<span class="o">(</span><span class="k">val</span> <span class="n">serviceId</span><span class="k">:</span> <span class="kt">String</span><span class="o">,</span> <span class="k">val</span> <span class="n">log</span><span class="k">:</span> <span class="kt">ActorRef</span><span class="o">)</span>
<span class="o">(</span><span class="k">implicit</span> <span class="k">val</span> <span class="n">system</span><span class="k">:</span> <span class="kt">ActorSystem</span><span class="o">,</span> <span class="k">val</span> <span class="n">ops</span><span class="k">:</span> <span class="kt">CRDTServiceOps</span><span class="o">[</span><span class="kt">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span>, <span class="kt">Set</span><span class="o">[</span><span class="kt">A</span><span class="o">]])</span>
<span class="k">extends</span> <span class="nc">CRDTService</span><span class="o">[</span><span class="kt">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span>, <span class="kt">Set</span><span class="o">[</span><span class="kt">A</span><span class="o">]]</span> <span class="o">{</span>
<span class="k">def</span> <span class="n">assign</span><span class="o">(</span><span class="n">id</span><span class="k">:</span> <span class="kt">String</span><span class="o">,</span> <span class="n">value</span><span class="k">:</span> <span class="kt">A</span><span class="o">)</span><span class="k">:</span> <span class="kt">Future</span><span class="o">[</span><span class="kt">Set</span><span class="o">[</span><span class="kt">A</span><span class="o">]]</span> <span class="k">=</span> <span class="n">op</span><span class="o">(</span><span class="n">id</span><span class="o">,</span> <span class="nc">AssignOp</span><span class="o">(</span><span class="n">value</span><span class="o">))</span>
<span class="o">}</span>
<span class="k">case</span> <span class="k">class</span> <span class="nc">AssignOp</span><span class="o">(</span><span class="n">value</span><span class="k">:</span> <span class="kt">Any</span><span class="o">)</span>
</code></pre></div></div>
<p>The <code class="highlighter-rouge">assign</code> service method creates an <code class="highlighter-rouge">AssignOp</code> instance, representing a persistent <em>assign</em> operation, that is first used by the local replica during the <em>prepare</em> phase, then written to the event log and finally used by the local and all remote replicas during the <em>effect</em> phase. <code class="highlighter-rouge">CRDTService[MVRegister[A], Set[A]]</code> also provides a <code class="highlighter-rouge">value(id: String): Future[Set[A]]</code> method for obtaining the current value of a local <code class="highlighter-rouge">MVRegister[A]</code> replica (usage shown in next section).</p>
<p>For being able to work, <code class="highlighter-rouge">MVRegisterService[A]</code> requires an implicit <code class="highlighter-rouge">CRDTServiceOps[MVRegister[A], Set[A]]</code> instance in scope. <a href="http://rbmhtechnology.github.io/eventuate/latest/api/index.html#com.rbmhtechnology.eventuate.crdt.CRDTServiceOps">CRDTServiceOps</a> is a type class that maps the CmRDT update phases <em>prepare</em> and <em>effect</em> to methods on CmRDT instances. It also defines how to obtain the current value from a CmRDT instance.</p>
<div class="language-scala highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.DurableEvent</span>
<span class="k">object</span> <span class="nc">MVRegister</span> <span class="o">{</span>
<span class="k">implicit</span> <span class="k">def</span> <span class="nc">MVRegisterServiceOps</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">CRDTServiceOps</span><span class="o">[</span><span class="kt">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span>, <span class="kt">Set</span><span class="o">[</span><span class="kt">A</span><span class="o">]]</span> <span class="o">{</span>
<span class="k">override</span> <span class="k">def</span> <span class="n">value</span><span class="o">(</span><span class="n">crdt</span><span class="k">:</span> <span class="kt">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">])</span><span class="k">:</span> <span class="kt">Set</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span> <span class="k">=</span>
<span class="n">crdt</span><span class="o">.</span><span class="n">value</span>
<span class="k">override</span> <span class="k">def</span> <span class="n">prepare</span><span class="o">(</span><span class="n">crdt</span><span class="k">:</span> <span class="kt">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">],</span> <span class="n">operation</span><span class="k">:</span> <span class="kt">Any</span><span class="o">)</span><span class="k">:</span> <span class="kt">Option</span><span class="o">[</span><span class="kt">Any</span><span class="o">]</span> <span class="k">=</span>
<span class="k">super</span><span class="o">.</span><span class="n">prepare</span><span class="o">(</span><span class="n">crdt</span><span class="o">,</span> <span class="n">operation</span><span class="o">)</span>
<span class="k">override</span> <span class="k">def</span> <span class="n">effect</span><span class="o">(</span><span class="n">crdt</span><span class="k">:</span> <span class="kt">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">],</span> <span class="n">operation</span><span class="k">:</span> <span class="kt">Any</span><span class="o">,</span> <span class="n">event</span><span class="k">:</span> <span class="kt">DurableEvent</span><span class="o">)</span><span class="k">:</span> <span class="kt">MVRegister</span><span class="o">[</span><span class="kt">A</span><span class="o">]</span> <span class="k">=</span>
<span class="n">operation</span> <span class="k">match</span> <span class="o">{</span>
<span class="k">case</span> <span class="nc">AssignOp</span><span class="o">(</span><span class="n">value</span><span class="o">)</span> <span class="k">=&gt;</span> <span class="n">crdt</span><span class="o">.</span><span class="n">assign</span><span class="o">(</span><span class="n">value</span><span class="o">.</span><span class="n">asInstanceOf</span><span class="o">[</span><span class="kt">A</span><span class="o">],</span> <span class="n">event</span><span class="o">.</span><span class="n">vectorTimestamp</span><span class="o">)</span>
<span class="o">}</span>
<span class="o">}</span>
<span class="o">}</span>
</code></pre></div></div>
<p>For <code class="highlighter-rouge">MVRegister[A]</code>, the <code class="highlighter-rouge">prepare</code> phase is a no-op and the default implementation from <code class="highlighter-rouge">super</code> should be used (or better, inherited). The <code class="highlighter-rouge">effect</code> phase uses the persistent <code class="highlighter-rouge">AssignOp</code> to call the <code class="highlighter-rouge">assign</code> method on the <code class="highlighter-rouge">MVRegister[A]</code> instance. The vector timestamp is taken from the <code class="highlighter-rouge">event</code> parameter that provides all event metadata.</p>
<p>The <a href="https://github.com/RBMHTechnology/eventuate/blob/master/eventuate-crdt/src/main/scala/com/rbmhtechnology/eventuate/crdt/MVRegister.scala">MVRegister</a> CmRDT and its service are defined in the <a href="http://rbmhtechnology.github.io/eventuate/download.html#id2">eventuate-crdt</a> module. Other CmRDTs provided by Eventuate are <a href="https://github.com/RBMHTechnology/eventuate/blob/master/eventuate-crdt/src/main/scala/com/rbmhtechnology/eventuate/crdt/Counter.scala">Counter</a>, <a href="https://github.com/RBMHTechnology/eventuate/blob/master/eventuate-crdt/src/main/scala/com/rbmhtechnology/eventuate/crdt/LWWRegister.scala">LWW-Register</a>, <a href="https://github.com/RBMHTechnology/eventuate/blob/master/eventuate-crdt/src/main/scala/com/rbmhtechnology/eventuate/crdt/ORSet.scala">OR-Set</a> and <a href="https://github.com/RBMHTechnology/eventuate/blob/master/eventuate-crdt/src/main/scala/com/rbmhtechnology/eventuate/crdt/ORCart.scala">OR-Cart</a> (an OR-Set based shopping cart CmRDT).</p>
<h3 id="crdt-service-usage">CRDT service usage</h3>
<p>The following example (full source code <a href="https://github.com/krasserm/eventuate-crdt-example/blob/master/src/main/scala/example/crdt/MVRegisterExample.scala">here</a>) sets up three locations, each running a local <code class="highlighter-rouge">MVRegisterService[String]</code> instance using the local event log for persisting <em>assign</em> operations. The local event logs are connected to each other via their <code class="highlighter-rouge">ReplicationEndpoint</code>s to form a replicated event log. This implements the reliable causal broadcast needed to disseminate <em>assign</em> operations to all replicas. The locations run all in the same JVM, each having its own <code class="highlighter-rouge">ActorSystem</code> (usually, locations run on different nodes in the same or even different data centers).</p>
<div class="language-scala highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="k">import</span> <span class="nn">akka.actor.ActorSystem</span>
<span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.ReplicationConnection</span>
<span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.ReplicationEndpoint</span>
<span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.crdt.MVRegisterService</span>
<span class="k">import</span> <span class="nn">com.rbmhtechnology.eventuate.log.leveldb.LeveldbEventLog</span>
<span class="k">import</span> <span class="nn">com.typesafe.config.ConfigFactory</span>
<span class="k">def</span> <span class="n">config</span><span class="o">(</span><span class="n">port</span><span class="k">:</span> <span class="kt">Int</span><span class="o">)</span><span class="k">:</span> <span class="kt">String</span> <span class="o">=</span>
<span class="c1">// set port in akka-remote config (omitted)
</span>
<span class="k">def</span> <span class="n">service</span><span class="o">(</span><span class="n">locationId</span><span class="k">:</span> <span class="kt">String</span><span class="o">,</span> <span class="n">port</span><span class="k">:</span> <span class="kt">Int</span><span class="o">,</span> <span class="n">connectToPorts</span><span class="k">:</span> <span class="kt">Set</span><span class="o">[</span><span class="kt">Int</span><span class="o">])</span><span class="k">:</span> <span class="kt">MVRegisterService</span><span class="o">[</span><span class="kt">String</span><span class="o">]</span> <span class="k">=</span> <span class="o">{</span>
<span class="k">implicit</span> <span class="k">val</span> <span class="n">system</span><span class="k">:</span> <span class="kt">ActorSystem</span> <span class="o">=</span>
<span class="nc">ActorSystem</span><span class="o">(</span><span class="nc">ReplicationConnection</span><span class="o">.</span><span class="nc">DefaultRemoteSystemName</span><span class="o">,</span> <span class="nc">ConfigFactory</span><span class="o">.</span><span class="n">parseString</span><span class="o">(</span><span class="n">config</span><span class="o">(</span><span class="n">port</span><span class="o">)))</span>
<span class="k">val</span> <span class="n">logName</span> <span class="k">=</span> <span class="s">"L"</span>
<span class="k">val</span> <span class="n">endpoint</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">ReplicationEndpoint</span><span class="o">(</span><span class="n">id</span> <span class="k">=</span> <span class="n">locationId</span><span class="o">,</span> <span class="n">logNames</span> <span class="k">=</span> <span class="nc">Set</span><span class="o">(</span><span class="n">logName</span><span class="o">),</span>
<span class="n">logFactory</span> <span class="k">=</span> <span class="n">logId</span> <span class="k">=&gt;</span> <span class="nc">LeveldbEventLog</span><span class="o">.</span><span class="n">props</span><span class="o">(</span><span class="n">logId</span><span class="o">),</span>
<span class="n">connections</span> <span class="k">=</span> <span class="n">connectToPorts</span><span class="o">.</span><span class="n">map</span><span class="o">(</span><span class="nc">ReplicationConnection</span><span class="o">(</span><span class="s">"127.0.0.1"</span><span class="o">,</span> <span class="k">_</span><span class="o">)))</span>
<span class="n">endpoint</span><span class="o">.</span><span class="n">activate</span><span class="o">()</span>
<span class="k">new</span> <span class="nc">MVRegisterService</span><span class="o">[</span><span class="kt">String</span><span class="o">](</span><span class="n">s</span><span class="s">"service-$locationId"</span><span class="o">,</span> <span class="n">endpoint</span><span class="o">.</span><span class="n">logs</span><span class="o">(</span><span class="n">logName</span><span class="o">))</span>
<span class="o">}</span>
<span class="k">val</span> <span class="n">serviceA</span> <span class="k">=</span> <span class="n">service</span><span class="o">(</span><span class="s">"A"</span><span class="o">,</span> <span class="mi">2552</span><span class="o">,</span> <span class="nc">Set</span><span class="o">(</span><span class="mi">2553</span><span class="o">,</span> <span class="mi">2554</span><span class="o">))</span> <span class="cm">/* at location A */</span>
<span class="k">val</span> <span class="n">serviceB</span> <span class="k">=</span> <span class="n">service</span><span class="o">(</span><span class="s">"B"</span><span class="o">,</span> <span class="mi">2553</span><span class="o">,</span> <span class="nc">Set</span><span class="o">(</span><span class="mi">2552</span><span class="o">,</span> <span class="mi">2554</span><span class="o">))</span> <span class="cm">/* at location B */</span>
<span class="k">val</span> <span class="n">serviceC</span> <span class="k">=</span> <span class="n">service</span><span class="o">(</span><span class="s">"C"</span><span class="o">,</span> <span class="mi">2554</span><span class="o">,</span> <span class="nc">Set</span><span class="o">(</span><span class="mi">2552</span><span class="o">,</span> <span class="mi">2553</span><span class="o">))</span> <span class="cm">/* at location C */</span>
<span class="c1">// ...
</span></code></pre></div></div>
<p>We will use a single <code class="highlighter-rouge">MVRegister[String]</code> instance, identified by <code class="highlighter-rouge">crdtId</code>, with a replica at each location. When calling <code class="highlighter-rouge">assign</code> or <code class="highlighter-rouge">value</code> for the first time after service creation, the service tries to recover the local replica from the event log by replaying operations or initializes a new one if there are no persistent operations for that instance id.</p>
<div class="language-scala highlighter-rouge"><div class="highlight"><pre class="highlight"><code><span class="k">val</span> <span class="n">crdtId</span> <span class="k">=</span> <span class="s">"1"</span>
<span class="n">serviceA</span><span class="o">.</span><span class="n">assign</span><span class="o">(</span><span class="n">crdtId</span><span class="o">,</span> <span class="s">"abc"</span><span class="o">).</span><span class="n">onSuccess</span> <span class="o">{</span>
<span class="k">case</span> <span class="n">r</span> <span class="k">=&gt;</span> <span class="n">println</span><span class="o">(</span><span class="n">s</span><span class="s">"assign result to replica at location A: $r"</span><span class="o">)</span>
<span class="o">}</span>
<span class="n">serviceB</span><span class="o">.</span><span class="n">assign</span><span class="o">(</span><span class="n">crdtId</span><span class="o">,</span> <span class="s">"xyz"</span><span class="o">).</span><span class="n">onSuccess</span> <span class="o">{</span>
<span class="k">case</span> <span class="n">r</span> <span class="k">=&gt;</span> <span class="n">println</span><span class="o">(</span><span class="n">s</span><span class="s">"assign result to replica at location B: $r"</span><span class="o">)</span>
<span class="o">}</span>
<span class="c1">// wait a bit ...
</span>
<span class="nc">Thread</span><span class="o">.</span><span class="n">sleep</span><span class="o">(</span><span class="mi">1000</span><span class="o">)</span>
<span class="n">serviceC</span><span class="o">.</span><span class="n">value</span><span class="o">(</span><span class="n">crdtId</span><span class="o">).</span><span class="n">onSuccess</span> <span class="o">{</span>
<span class="k">case</span> <span class="n">r</span> <span class="k">=&gt;</span> <span class="n">println</span><span class="o">(</span><span class="n">s</span><span class="s">"read result from replica at location C: $r"</span><span class="o">)</span>
<span class="o">}</span>
</code></pre></div></div>
<p>The updates start when <code class="highlighter-rouge">serviceA</code> assigns its local replica the value <code class="highlighter-rouge">abc</code> and <code class="highlighter-rouge">serviceB</code> its local replica the value <code class="highlighter-rouge">xyz</code>. These operations are expected to be concurrent because <code class="highlighter-rouge">serviceB</code> will likely have finished the local update before having received the <em>assign</em> operation from <code class="highlighter-rouge">serviceA</code>. However, there is a small chance that this is not the case. We can find that out by reading the value from <code class="highlighter-rouge">serviceC</code>, after having given operation dissemination enough time. In case of concurrent operations, output similar to the following is generated:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>assign result to replica at location A: Set(abc)
assign result to replica at location B: Set(xyz)
read result from replica at location C: Set(abc, xyz)
</code></pre></div></div>
<p>If the read value from <code class="highlighter-rouge">serviceC</code> contains both assignments, the operations were concurrent. If it only contains a single assignment, the operations are causally related.</p>
<h2 id="production-deployment-considerations">Production deployment considerations</h2>
<p>All operations executed on a single CRDT service instance go to the same event log which puts an upper limit on write throughput. Applications that want to scale writes should consider creating <a href="http://rbmhtechnology.github.io/eventuate/reference/event-log.html#replicated-event-log">multiple replicated event logs</a> and partition CRDT service instances accordingly. On the other hand, applications with moderate write rates may also share the same event log among multiple CRDT services of different type. Eventuate properly isolates these services.</p>
<p>As long as the total number of updates per CRDT instance is moderate e.g. less than a few thousand updates it is not necessary to create snapshots of CRDT state. Eventuate event logs are indexed on CRDT id (i.e. <code class="highlighter-rouge">aggregateId</code>) so that operation replay per instance is reasonably fast. For a much larger number of updates per instance it is recommended to save snapshots. CRDT services provide a generic <code class="highlighter-rouge">save(id: String): Future[SnapshotMetadata]</code> method for saving CRDT state snapshots where <code class="highlighter-rouge">id</code> is the CRDT instance id.</p>
<p>All CRDT services provided by Eventuate serialize CRDT operations and snapshots with Google <a href="https://developers.google.com/protocol-buffers/">protocol buffers</a>. The corresponding serializers are configured as described in <a href="http://rbmhtechnology.github.io/eventuate/reference/event-sourcing.html#custom-serialization">Custom serialization</a>. When using the CRDT framework for the development of custom CRDT services, it is recommended to provide custom serializers as well, otherwise Akka’s default serializer i.e. Java serializer is used for serializing operations and snapshots.</p>
<p>Eventuate’s CRDTs and its reliable causal broadcast infrastructure have been heavily <a href="https://github.com/RBMHTechnology/eventuate-chaos/">tested under chaotic conditions</a> with all currently available <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#storage-backends">storage backends</a>. Chaos was generated by randomly injecting network partitions between Eventuate locations, within Cassandra clusters and between Eventuate locations and Cassandra cluster nodes. Furthermore, network packets have been dropped randomly too.</p>
<p>In all tests, replicated state successfully converged at all locations, even after long-running chaos tests (approx. 1 hour). A single missing event or a duplicate delivered to in-memory CRDT replicas would have broken convergence and let the chaos tests fail. The failure handling consequences learned during the chaos tests are summarized in the <a href="http://rbmhtechnology.github.io/eventuate/reference/event-sourcing.html#failure-handling">Failure handling</a> section of the Eventuate documentation.</p>
<h2 id="planned-features">Planned features</h2>
<p>In addition to the basic CRDTs that are currently part of Eventuate, we plan to work on tree and graph CRDTs as well as CRDTs for collaborative text editing. We are also working on a Java API for the framework. A Java API for the existing CRDT services <a href="https://github.com/RBMHTechnology/eventuate/tree/master/eventuate-crdt/src/main/scala/com/rbmhtechnology/eventuate/crdt/japi">is already available</a>.</p>
<p>Eventuate’s CRDTs currently follow the specifications in <a href="http://hal.upmc.fr/file/index/docid/555588/filename/techreport.pdf">[1]</a>. As explained in <a href="http://haslab.uminho.pt/sites/default/files/ashoker/files/opbaseddais14.pdf">[2]</a> these specifications can be improved to make them <em>pure</em> operation-based. A prerequisite for such an implementation is a <em>tagged</em> reliable causal broadcast (TRCB) which is a RCB middleware that additionally exposes causality metadata to the application. Eventuate already delivers vector timestamps together with events to applications and can therefore be used as TRCB. We currently think about factoring out the TRCB part of Eventuate into a separate open source project.</p>
<p>Switching to pure operation-based CRDTs also increases write throughput as explained in <a href="https://github.com/RBMHTechnology/eventuate/issues/301">ticket 301</a>. We also work on increasing replication throughput by parallelizing stages in the replication pipelines and reducing the replication protocol overhead.</p>
<h2 id="references">References</h2>
<p>[1] <a href="http://hal.upmc.fr/file/index/docid/555588/filename/techreport.pdf">A comprehensive study of Convergent and Commutative Replicated Data Types</a>, 2011, Marc Shapiro et. al.<br />
[2] <a href="http://haslab.uminho.pt/sites/default/files/ashoker/files/opbaseddais14.pdf">Making Operation-based CRDTs Operation-based</a>, 2014, Carlos Baquero et. al.</p>
Wed, 19 Oct 2016 00:00:00 +0000http://krasserm.github.io/2016/10/19/operation-based-crdt-framework/
http://krasserm.github.io/2016/10/19/operation-based-crdt-framework/Chaos testing with Docker and Cassandra on Mac OS X<hr />
<p><strong>12.12.2015</strong>: This blog posts refers to an <a href="https://github.com/RBMHTechnology/eventuate-chaos/tree/blog-01">initial version</a> of the <a href="https://github.com/RBMHTechnology/eventuate-chaos">eventuate-chaos</a> tools project. Meanwhile, this project has been significantly improved by <a href="https://github.com/kongo2002">Gregor Uhlenheuer</a> and provides now a rich and user-friendly tool set for injecting random network and node failures into <a href="http://cassandra.apache.org/">Apache Cassandra</a> and <a href="https://github.com/RBMHTechnology/eventuate">Eventuate</a> clusters.</p>
<hr />
<p>In this blog post I summarize the setup of a local <a href="http://cassandra.apache.org/">Cassandra</a> cluster with <a href="https://www.docker.com/">Docker</a> on Mac OS X for chaos testing on a single machine. Chaos is generated by a coordinator written in Scala that randomly kills and restarts cluster nodes. I used <a href="http://boot2docker.io/">boot2docker</a> 1.6.2 and Mac OS X 10.9.5 but the following should also work with newer versions. It is assumed that the boot2docker VM is running:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ boot2docker start
Waiting for VM and Docker daemon to start...
..........ooooooooo
Started.
</code></pre></div></div>
<p>and Mac OS terminal sessions are initialized with:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ eval `boot2docker shellinit`
</code></pre></div></div>
<h2 id="running-the-cluster">Running the cluster</h2>
<p>There are several Cassandra Docker images available. The <a href="https://registry.hub.docker.com/_/cassandra/">image used here</a> is that from the <a href="https://github.com/docker-library/official-images">Docker Official Images</a> project. For starting a three node Cassandra 2.1.6 cluster, a seed node is started first:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ docker run --name cassandra-1 -d cassandra:2.1.6
b647c51ba090cf66fc83919c7918ea827ff28faf3856e484b8fc703b4f520f82
</code></pre></div></div>
<p>The IP address of the seed node container can be obtained with:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ SEED=`docker inspect --format='{{ .NetworkSettings.IPAddress }}' cassandra-1`
almdudler:~ martin$ echo $SEED
127.17.0.1
</code></pre></div></div>
<p>The <code class="highlighter-rouge">SEED</code> value is used for starting the other two nodes:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ docker run --name cassandra-2 -d -e CASSANDRA_SEEDS=$SEED cassandra:2.1.6
4ed90829a438670c8a496681f65078b23d5609f5ea55bead915ed07fdcb30ba5
almdudler:~ martin$ docker run --name cassandra-3 -d -e CASSANDRA_SEEDS=$SEED cassandra:2.1.6
9dfef16988f5950cb21523748cb294f0651ed0781592afe04948abb3044ed186
</code></pre></div></div>
<p>That’s it, the cluster is up and running. It can be stopped with:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ docker stop cassandra-1 cassandra-2 cassandra-3
cassandra-1
cassandra-2
cassandra-3
</code></pre></div></div>
<p>The containers (and all stored data) can be removed with:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ docker rm cassandra-1 cassandra-2 cassandra-3
cassandra-1
cassandra-2
cassandra-3
</code></pre></div></div>
<p>I also started to work on a small <a href="https://github.com/RBMHTechnology/eventuate-chaos/tree/blog-01">utilities project</a> that provides scripts for starting and stopping Cassandra clusters. For example, starting a four node cluster is as simple as:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:eventuate-chaos martin$ ./cluster-start.sh 4
cassandra-1
cassandra-2
cassandra-3
cassandra-4
</code></pre></div></div>
<p>The script internally uses the above <code class="highlighter-rouge">docker</code> commands for starting and stopping nodes. The cluster can be stopped and the containers removed with:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:eventuate-chaos martin$ ./cluster-stop.sh
cassandra-1
cassandra-2
cassandra-3
cassandra-4
</code></pre></div></div>
<p>Find more details in project’s <a href="https://github.com/RBMHTechnology/eventuate-chaos/blob/blog-01/README.md#manual-failure-generation">README</a>.</p>
<h2 id="accessing-the-nodes">Accessing the nodes</h2>
<p>The started containers are not directly accessible from Mac OS because they are running within the boot2docker VM. The goal however is to access these containers directly from Mac OS. This can be achieved with <em>port mapping</em> or <em>custom routing</em>.</p>
<h3 id="port-mapping">Port mapping</h3>
<p>With port mapping, container ports are mapped to boot2docker ports so that applications that are running on Mac OS can connect to the boot2docker VM. For example, when starting the seed node with</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>docker run --name cassandra-1 -d -p 9042:9042 cassandra:2.1.6
</code></pre></div></div>
<p>a connection to that node can be established with the boot2docker IP address:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ telnet `boot2docker ip` 9042
Trying 192.168.59.103...
Connected to 192.168.59.103.
</code></pre></div></div>
<p>Port mapping works well for tools like <code class="highlighter-rouge">cqlsh</code> but can be problematic when using the <a href="https://github.com/datastax/java-driver">Datastax Java Driver</a>, for example: nodes that are added to the cluster are advertised to the driver with their container IP addresses which cannot be accessed if the driver is running directly on Mac OS. In this case, a custom route can be added to the Mac OS routing tables that routes all traffic targeted at docker containers to the boot2docker VM, as shown in the next section.</p>
<h3 id="custom-routing">Custom routing</h3>
<p>Assuming that Docker containers have IP addresses <code class="highlighter-rouge">172.17.x.x</code>, a custom route to the boot2docker VM can be added with:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ sudo route -n add 172.17.0.0/16 `boot2docker ip`
add net 172.17.0.0: gateway 192.168.59.103
</code></pre></div></div>
<p>A <code class="highlighter-rouge">netstat -nr</code> output should now contain something like:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ netstat -nr
Routing tables
Internet:
Destination Gateway Flags Refs Use Netif Expire
...
172.17 192.168.59.103 UGSc 0 0 vboxnet
...
</code></pre></div></div>
<p>Docker containers are now directly accessible from Mac OS. Assuming a running seed node container with IP address <code class="highlighter-rouge">172.17.0.1</code>, a <code class="highlighter-rouge">telnet</code> client should be able to connect:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:~ martin$ telnet 172.17.0.1 9042
Trying 172.17.0.1...
Connected to 172.17.0.1.
</code></pre></div></div>
<h2 id="generating-chaos">Generating chaos</h2>
<p>For randomly stopping and restarting nodes, the <a href="https://github.com/RBMHTechnology/eventuate-chaos/tree/blog-01">utilities project</a> provides a coordinator application named <a href="https://github.com/RBMHTechnology/eventuate-chaos/blob/blog-01/src/main/scala/com/rbmhtechnology/eventuate/chaos/ChaosCluster.scala"><code class="highlighter-rouge">ChaosCluster</code></a> which can be started from <a href="http://www.scala-sbt.org/">sbt</a> and configured with the parameters defined in <a href="https://github.com/RBMHTechnology/eventuate-chaos/blob/blog-01/src/main/resources/reference.conf">reference.conf</a>. Running <code class="highlighter-rouge">ChaosCluster</code> with default settings first starts a four node cluster:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>almdudler:eventuate-chaos martin$ sbt
[info] Loading global plugins from /Users/martin/.sbt/0.13/plugins
[info] Loading project definition from /Users/martin/eventuate-chaos/project
[info] Set current project to eventuate-chaos (in build file:/Users/martin/eventuate-chaos/)
&gt; runMain com.rbmhtechnology.eventuate.chaos.ChaosCluster
[info] Running com.rbmhtechnology.eventuate.chaos.ChaosCluster
Writing /Users/martin/.boot2docker/certs/boot2docker-vm/ca.pem
Writing /Users/martin/.boot2docker/certs/boot2docker-vm/cert.pem
Writing /Users/martin/.boot2docker/certs/boot2docker-vm/key.pem
cassandra-1
cassandra-2
cassandra-3
cassandra-4
Cluster started. Press any key to start chaos ...
</code></pre></div></div>
<p>Pressing any key, after the cluster started, generates chaos by randomly stopping and (re)starting nodes, except the seed node (<code class="highlighter-rouge">cassandra-1</code>):</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>cassandra-4
cassandra-2
Node(s) stopped. Press any key to stop cluster ...
cassandra-4
cassandra-2
Node(s) started. Press any key to stop cluster ...
cassandra-3
Node(s) stopped. Press any key to stop cluster ...
cassandra-3
Node(s) started. Press any key to stop cluster ...
</code></pre></div></div>
<p>Pressing any key a second time stops the cluster and removes all containers:</p>
<div class="highlighter-rouge"><div class="highlight"><pre class="highlight"><code>cassandra-1
cassandra-2
cassandra-3
cassandra-4
Cluster stopped
[success] Total time: 77 s, completed 11.07.2015 17:12:22
</code></pre></div></div>
<p>The <code class="highlighter-rouge">ChaosCluster</code> application uses the <a href="http://www.scala-lang.org/api/current/index.html#scala.sys.process.package"><code class="highlighter-rouge">scala.sys.process</code></a> API for executing <code class="highlighter-rouge">docker</code> commands (see <a href="https://github.com/RBMHTechnology/eventuate-chaos/blob/blog-01/src/main/scala/com/rbmhtechnology/eventuate/chaos/ChaosCommands.scala"><code class="highlighter-rouge">ChaosCommands</code></a> trait for details). Later versions of the project will additionally provide utilities for running distributed Eventuate applications in Docker containers that are also subject to random failures.</p>
Mon, 13 Jul 2015 00:00:00 +0000http://krasserm.github.io/2015/07/13/chaos-testing-with-docker-and-cassandra/
http://krasserm.github.io/2015/07/13/chaos-testing-with-docker-and-cassandra/A comparison of Akka Persistence with Eventuate<hr />
<p><strong>23.10.2015</strong>: Several updates in all sections to cover recent questions and developments.</p>
<hr />
<p>The following is an attempt to describe the similarities and differences between <a href="http://doc.akka.io/docs/akka/2.4.0/scala/persistence.html">Akka Persistence</a> and <a href="http://rbmhtechnology.github.io/eventuate/">Eventuate</a>. Both are Akka-based <a href="http://martinfowler.com/eaaDev/EventSourcing.html">event-sourcing</a> and <a href="http://martinfowler.com/bliki/CQRS.html">CQRS</a> tools written in Scala, making different distributed system compromises. For an introduction to these tools, please take a look at their online documentation.</p>
<p>I’m the original author of both, Akka Persistence and Eventuate, currently focusing exclusively on the development of Eventuate. Of course, I’m totally biased ;) Seriously, if I have goofed something, please let me know.</p>
<h2 id="command-side">Command side</h2>
<p>In Akka Persistence, the command side (C of CQRS) is represented by <code class="highlighter-rouge">PersistentActor</code>s (PAs), in Eventuate by <code class="highlighter-rouge">EventsourcedActor</code>s (EAs). Their internal state represents an application’s write model.</p>
<p>PAs and EAs validate new commands against the write model and if validation succeeds, generate and persist one or more events which are then handled to update internal state. After a crash or a normal application re-start, internal state is recovered by replaying persisted events from the event log, optionally starting from a snapshot. PAs and EAs also support sending messages with at-least-once delivery semantics to other actors. Akka Persistence provides the <code class="highlighter-rouge">AtLeastOnceDelivery</code> trait for that purpose, Eventuate the <code class="highlighter-rouge">ConfirmedDelivery</code> trait.</p>
<p>From this perspective PAs and EAs are quite similar. A major difference is that PAs must be singletons whereas EAs can be replicated and updated concurrently. If an Akka Persistence application accidentally creates and updates two PA instances with the same <code class="highlighter-rouge">persistenceId</code>, the underlying event log will be corrupted, either by overwriting existing events or by appending conflicting events. Akka Persistence event logs are designed for having only a single writer and cannot be shared.</p>
<p>In Eventuate, EAs can share an event log. Events emitted by one EA can be consumed by other EAs, based on predefined and customizable event routing rules. In other words, EAs can collaborate by exchanging events over a shared event log. This collaboration can be, for example, a distributed business process executed by EAs of different type, or state replication where EAs of the same type reconstruct and update internal state at multiple locations. These locations can even be globally distributed. Event replication between locations is asynchronous and reliable.</p>
<h2 id="event-relations">Event relations</h2>
<p>In Akka Persistence, events have a total order per PA but events emitted by different PAs are not related. Even if an event emitted by one PA actually happened before an event emitted by another PA, this relationship is not tracked by Akka Persistence. For example, if PA<sub>1</sub> persists an event e<sub>1</sub>, then sends a command to PA<sub>2</sub> which in turn persists another event e<sub>2</sub> during handling of that command, e<sub>1</sub> obviously happened before e<sub>2</sub> but applications cannot determine this relationship by comparing e<sub>1</sub> with e<sub>2</sub>.</p>
<p>Eventuate additionally tracks the happened-before relationship (= potential causality) of events. For example, if EA<sub>1</sub> persists an event e<sub>1</sub> and EA<sub>2</sub> persists an event e<sub>2</sub> after having consumed e<sub>1</sub>, then e<sub>1</sub> happened before e<sub>2</sub> which is also tracked. Happened-before relationships are tracked with <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#vector-clocks">vector clocks</a> and applications can determine whether any two events have a happened-before relationship or are concurrent by comparing their vector timestamps.</p>
<p>Tracking the happened-before relationship of events is a prerequisite for running multiple replicas of an EA. An EA that consumes events from its replicas must be able to determine whether its last internal state update happened before or is concurrent to (and potentially in conflict with) the consumed events.</p>
<p>If the last internal state update happened before a consumed event, that event can be handled as regular update. If it is concurrent to the consumed event, the event might be a conflicting event and must be handled accordingly. If an EA’s internal state is a <a href="http://en.wikipedia.org/wiki/Conflict-free_replicated_data_type">CRDT</a>, for example, the conflict can be resolved automatically (see also Eventuate’s <a href="http://rbmhtechnology.github.io/eventuate/user-guide.html#operation-based-crdts">operation-based CRDTs</a>). If internal state is not a CRDT, Eventuate provides further means to <a href="http://rbmhtechnology.github.io/eventuate/user-guide.html#tracking-conflicting-versions">track</a> and <a href="http://rbmhtechnology.github.io/eventuate/user-guide.html#resolving-conflicting-versions">resolve</a> conflicts, either automatically or interactively.</p>
<h2 id="event-logs">Event logs</h2>
<p>As already mentioned, in Akka Persistence each PA has its own private event log. Depending on the storage backend, an event log is either stored redundantly on several nodes (e.g. synchronously replicated for stronger durability guarantees) or stored locally. In either case, Akka Persistence requires a strongly consistent view on an event log.</p>
<p>For example, a PA that crashed and recovers on another node must be able to read all previously written events in correct order, otherwise recovery may be incomplete and the PA may later overwrite existing events or append new events to the log that are in conflict with existing but unread events. Therefore, only storage backends that support strong consistency can be used for Akka Persistence.</p>
<p>The write availability of an Akka Persistence application is constrained by the write availability of the underlying storage backend. According to the <a href="http://en.wikipedia.org/wiki/CAP_theorem">CAP theorem</a>, write availability of a strongly consistent, distributed storage backed is limited. Consequently, the command side of an Akka Persistence application chooses CP from CAP.</p>
<p>These constraints make it difficult to globally distribute an Akka Persistence application as strong consistency and total event ordering also require global coordination. Eventuate goes one step further here: it requires strong consistency and total event ordering only within a so called <em>location</em>. A location can be a data center, a (micro-)service, a node in a cluster or a process on a single node, to mention a few examples.</p>
<p>An Eventuate application that only consists of a single location implements the same consistency model as an Akka Persistence application. However, Eventuate applications usually consist of multiple locations. Events generated at individual locations are asynchronously and reliably replicated to other locations. Inter-location event replication is Eventuate-specific and preserves causal event storage order. Storage backends at different locations do not communicate directly with each other. Therefore, different storage backends can be used at different locations.</p>
<p>An Eventuate <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#event-logs">event log</a> that is replicated across locations is called a <em>replicated event log</em>, its representation at a given location is called a <em>local event log</em>. EAs deployed at different locations can exchange events by sharing a replicated event log. This allows for EA state replication across locations. EAs and their underlying event logs remain writeable even during inter-location network partitions. From this perspective, a multi-location Eventuate application chooses AP from CAP. Writes during a network partition at different locations may cause conflicts which can be resolved as described previously.</p>
<p>By introducing partition-tolerant locations, a global total ordering of events is not possible any more. The strongest partial ordering that is possible under these constraints is causal ordering i.e. an ordering that preserves the happened-before relation of events. In Eventuate, every location guarantees the delivery of events in causal order to their local EAs (and views, see <a href="#query-side">next section</a>). The delivery order of concurrent events may differ at individual locations but is repeatable within a given location.</p>
<h2 id="query-side">Query side</h2>
<p>In Akka Persistence, the query side (Q of CQRS) can be implemented with <code class="highlighter-rouge">PersistentView</code>s (PVs). A PV is currently limited to consume events from only one PA. This limitation has been <a href="https://groups.google.com/forum/#!msg/akka-user/MNDc9cVG1To/blqgyC7sIRgJ">intensively discussed</a> on the Akka mailing list. A proposed solution, available since Akka 2.4, is <a href="http://doc.akka.io/docs/akka/2.4.0/scala/persistence-query.html">Akka Persistence Query</a>: storage plugins may provide support for aggregating events from multiple PAs and serve the result as <a href="http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala.html">Akka Streams</a> <code class="highlighter-rouge">Source</code>.</p>
<p>In Eventuate, the query side can be implemented with <code class="highlighter-rouge">EventsourcedView</code>s (EVs). An EV can consume events from all EAs that share an event log, even if they are globally distributed. Events are always consumed in correct causal order. An application can either have a single replicated event log or several event logs, organized around topics, for example. Future extensions will allow EVs to consume events from multiple event logs. An Akka Streams API in Eventuate is also planned.</p>
<h2 id="storage-plugins">Storage plugins</h2>
<p>From a storage plugin perspective, events in Akka persistence are primarily organized around <code class="highlighter-rouge">persistenceId</code> i.e. around PA instances having their own private event log. Aggregating events from several PAs requires either the creation of an additional index in the storage backend or an on-the-fly event stream composition when serving a query. In Eventuate, events from several EAs are stored in the same shared event log. During recovery, EAs that don’t have an <code class="highlighter-rouge">aggregateId</code> defined, consume all events from the event log while those with a defined <code class="highlighter-rouge">aggregateId</code> only consume events with that <code class="highlighter-rouge">aggregateId</code> as routing destination. This requires Eventuate storage plugins to maintain a separate index from which events can be replayed by <code class="highlighter-rouge">aggregateId</code>.</p>
<p>Akka Persistence has a public storage plugin API for journals and snapshot stores with many implementations <a href="http://akka.io/community/">contributed by the community</a>. Eventuate will also define a public storage plugin API in the future. At the moment, applications can choose between a <a href="http://rbmhtechnology.github.io/eventuate/reference/event-log.html#leveldb-storage-backend">LevelDB storage backend</a> and a <a href="http://rbmhtechnology.github.io/eventuate/reference/event-log.html#cassandra-storage-backend">Cassandra storage backend</a>.</p>
<h2 id="throughput">Throughput</h2>
<p>Both, PAs in Akka Persistence and EAs in Eventuate can choose whether to keep internal state in sync with the event log. This is relevant for applications that need to validate new commands against internal state before persisting new events. To prevent validation against stale state, new commands must be delayed until a currently running event write operation successfully completed. PAs support this with a <code class="highlighter-rouge">persist</code> method (in contrast to <code class="highlighter-rouge">persistAsync</code>), EAs with a <code class="highlighter-rouge">stateSync</code> boolean property.</p>
<p>A consequence of synchronizing internal state with an event log is decreased throughput. Synchronizing internal state has a stronger impact in Akka Persistence than in Eventuate because of the details how event batch writes are implemented. In Akka Persistence, events are batched on PA level, but only when using <code class="highlighter-rouge">persistAsync</code>. In Eventuate there’s a separate batching layer between EAs and the storage plugin, so that events emitted by different EA instances, even if they sync their internal state with the event log, can be batched for writing.</p>
<p>Comparing the write throughput of two single PA and EA instances, they are approximately the same in Akka Persistence and Eventuate (assuming a comparable storage plugin). However, in Eventuate, the overall write throughput can increase with an increasing number of EA instances, whereas the write throughput in Akka Persistence can not. This is especially relevant for applications that follow a one PA/EA per <a href="http://martinfowler.com/bliki/DDD_Aggregate.html">aggregate</a> design with thousands to millions active (= writable) instances. Looking at the Akka Persistence code, I think it shouldn’t be too much effort moving the batching logic of PA down to a separate batching layer.</p>
<h2 id="conclusion">Conclusion</h2>
<p>Eventuate supports the same consistency model as Akka Persistence but additionally allows relaxation to causal consistency. This relaxation is necessary for EA high-availability and partition tolerance (AP of CAP). Eventuate also supports reliable actor collaboration based on causally ordered, de-duplicated event streams. From these perspectives, Eventuate is a functional superset of Akka Persistence.</p>
<p>Choosing availability over consistency requires that conflict detection and resolution (either automated or interactive) must be a primary concern. Eventuate supports that by providing operation-based CRDTs as well as utilities and APIs for tracking and resolving conflicting versions of application state.</p>
<p>Handling conflicts rather than preventing them is important for the resilience of distributed systems. Being able to remain operable within a location that is temporarily partitioned from other locations also makes Eventuate an interesting option for offline use cases.</p>
<p>Eventuate is still a young project. It started as a prototype in late 2014 and was <a href="https://github.com/RBMHTechnology/eventuate">open-sourced</a> in 2015. It is actively developed in context of the <a href="http://www.redbullmediahouse.com/">Red Bull Media House</a> (RBMH) <a href="http://rbmhtechnology.github.io/">Open Source Initiative</a> and primarily driven by internal RBMH projects.</p>
Mon, 25 May 2015 00:00:00 +0000http://krasserm.github.io/2015/05/25/akka-persistence-eventuate-comparison/
http://krasserm.github.io/2015/05/25/akka-persistence-eventuate-comparison/Event sourcing at global scale<p>Together with an <a href="http://www.redbullmediahouse.com/">international customer</a>, I recently started to explore several options how to globally distribute an application that is based on <a href="http://martinfowler.com/eaaDev/EventSourcing.html">event sourcing</a>. The main driver behind this initiative is the requirement that geographically distinct locations (called <em>sites</em>) shall have low-latency access to the application: each site shall run the application in a near data center and application data shall be replicated across all sites. A site shall also remain available for writes if there are inter-site network partitions. When a partition heals, updates from different sites must be merged and conflicts (if any) resolved.</p>
<p>In this blog post, I’ll briefly summarize our approach. We also validated our approach with a prototype that we <a href="https://github.com/RBMHTechnology/eventuate">recently open-sourced</a>. During this year, we’ll develop this prototype into a production-ready toolkit for event sourcing at global scale.</p>
<p>As a starting point for our prototype, we used <a href="http://doc.akka.io/docs/akka/2.3.8/scala/persistence.html">akka-persistence</a> but soon found that the conceptual and technical extensions we needed were quite hard to implement on top of the current version of akka-persistence (2.3.8). We therefore decided for a lightweight re-implementation of the akka-persistence API (with some modifications) together with our extensions for geo-replication. Of course, we are happy to contribute our work back to akka-persistence later, should there be broader interest in our approach.</p>
<p>The extensions we wrote are not only useful in context of geo-replication but can also be used to overcome some of the current limitations in akka-persistence. For example, in akka-persistence, event-sourced actors must be cluster-wide singletons. With our approach, we allow several instances of an event-sourced actor to be updated concurrently on multiple nodes and conflicts (if any) to be detected and resolved.
Also, we support event aggregation from several (even globally distributed) producers in a scalable way together with a deterministic replay of these events.</p>
<p>The following sections give an overview of the system model we developed. It is assumed that the reader is already familiar with the basics of <a href="http://martinfowler.com/eaaDev/EventSourcing.html">event sourcing</a>, <a href="http://martinfowler.com/bliki/CQRS.html">CQRS</a> and <a href="http://doc.akka.io/docs/akka/2.3.8/scala/persistence.html">akka-persistence</a>.</p>
<h2 id="sites">Sites</h2>
<p>In our model, a geo-replicated event-sourced application is distributed across <em>sites</em> where each site is operated by a separate data center. For low-latency access, users interact with a site that is geographically close.</p>
<p>Application events generated on one site are asynchronously replicated to other sites so that application state can be reconstructed on all sites from a site-local event log. A site remains available for writes even if it is partitioned from other sites.</p>
<h2 id="event-log">Event log</h2>
<p>At the system’s core is a globally replicated event log that preserves the happened-before relationship (= potential causal relationship) of events. Happened-before relationships are tracked with vector timestamps. They are generated by one or more <a href="http://en.wikipedia.org/wiki/Vector_clock">vector clocks</a> on each site and stored together with events in the event log. By comparing vector timestamps, one can determine whether any two events have a happened-before relationship or are concurrent.</p>
<p>The partial ordering of events, given by their vector timestamps, is preserved in each site-local copy of the replicated event log: if <code class="highlighter-rouge">e1 -&gt; e2</code> then <code class="highlighter-rouge">offset(e1) &lt; offset(e2)</code>, where <code class="highlighter-rouge">-&gt;</code> is the happened-before relation and <code class="highlighter-rouge">offset(e)</code> is the position or index of event <code class="highlighter-rouge">e</code> in a site-local event log. For example, if site A writes event <code class="highlighter-rouge">e1</code> that (when replicated) causes an event <code class="highlighter-rouge">e2</code> on site B, then the replication protocol ensures that <code class="highlighter-rouge">e1</code> is always stored before <code class="highlighter-rouge">e2</code> in <em>all</em> site-local event logs. As a direct consequence of that storage order, applications that produce to and consume from the event log experience event replication as reliable, causally ordered event multicast: if <code class="highlighter-rouge">emit(e1) -&gt; emit(e2)</code> then all applications on all sites will consume <code class="highlighter-rouge">e1</code> before <code class="highlighter-rouge">e2</code>, where <code class="highlighter-rouge">-&gt;</code> is the happened-before relation and <code class="highlighter-rouge">emit(e)</code> writes event <code class="highlighter-rouge">e</code> to the site-local event log.</p>
<p>The relative position of concurrent events in a site-local event log is not defined i.e. concurrent events may have a different ordering in different site-local event logs. Their replay, however, is deterministic per site, as a site-local event log imposes a total ordering on local event copies (which is helpful for debugging purposes, for example). A global total ordering of events is not an option in our case, as it would require global coordination which is in conflict with the availability requirement of partitioned sites. It would furthermore increase write latencies significantly.</p>
<p>In our implementation, we completely separate inter-site event replication from (optional) intra-site event replication. We use intra-site replication only for stronger durability guarantees i.e. for making a site-local event log highly available. We implemented asynchronous inter-site replication independent from concrete event storage backends such as <a href="https://github.com/google/leveldb">LevelDB</a>, <a href="http://kafka.apache.org/">Kafka</a>, <a href="http://cassandra.apache.org/">Cassandra</a> or whatever. This allows us to replace storage backends more easily whenever needed.</p>
<h2 id="event-sourced-actors">Event-sourced actors</h2>
<p>We distinguish two types of actors that interact with the event log: <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#event-sourced-actors"><code class="highlighter-rouge">EventsourcedActor</code></a> and <a href="http://rbmhtechnology.github.io/eventuate/architecture.html#event-sourced-views"><code class="highlighter-rouge">EventsourcedView</code></a>. They correspond to <code class="highlighter-rouge">PersistentActor</code> and <code class="highlighter-rouge">PersistentView</code> in akka-persistence, respectively, but with a major difference in event consumption.</p>
<p>Like in akka-persistence, <code class="highlighter-rouge">EventsourcedActor</code>s (EAs) produce events to the event log (during command processing) and consume events from the event log. A major difference is that EAs do not only consume events they produce themselves but also consume events that other EAs produce to the same event log (which can be customized by filter criteria). In other words, EAs do not only consume events to reconstruct internal state but also to collaborate with each other by exchanging events which is at the heart of <a href="http://en.wikipedia.org/wiki/Event-driven_architecture">event-driven architectures</a> and <a href="http://martinfowler.com/eaaDev/EventCollaboration.html">event collaboration</a>.</p>
<p>From this perspective, a replicated event log is the backbone of a distributed, durable and causality-preserving event bus that also provides the full history of events, so that event consumers can reconstruct application state any time by replaying events. For exchanging events, EAs may be co-located at the same site (Fig. 1) or distributed across sites (Fig. 2)</p>
<p><img src="/img/2015-01-13/intra-site.png" alt="Intra-site EA collaboration" title="Intra-site EA collaboration" />
Fig. 1: Intra-site EA collaboration</p>
<p><img src="/img/2015-01-13/inter-site.png" alt="Inter-site EA collaboration" title="Inter-site EA collaboration" />
Fig. 2: Inter-site EA collaboration</p>
<p>We think that our distributed event bus might be an interesting implementation option of Akka’s <a href="http://doc.akka.io/docs/akka/2.3.8/scala/event-bus.html">event bus</a>, especially for distributed event-based collaboration in an Akka <a href="http://doc.akka.io/docs/akka/2.3.8/scala/cluster-usage.html">cluster</a>. In this case, Akka cluster applications could also rely on causal ordering of events.</p>
<p>One special mode of collaboration is state replication: EA instances of the same type consume each other’s events to reconstruct application state on different sites (more on that later). A related example is to maintain hot-standby instances of EAs on the same site to achieve fail-over times of milliseconds. Another example of collaboration is a distributed business process: EAs of different type process each other’s events to achieve a common goal. Reliability of the distributed business process is given by durability of events in the event log and event replay in case of failures.</p>
<p>For sending messages to other non-event-sourced actors (external services, …), EAs have several options:</p>
<ul>
<li>during command processing with at-most-once message delivery semantics. The same option exists for <code class="highlighter-rouge">PersistentActor</code>s in akka-persistence by using a custom <a href="http://doc.akka.io/docs/akka/2.3.8/scala/persistence.html#event-sourcing"><code class="highlighter-rouge">persist</code></a> handler.</li>
<li>during event processing with at-least-once message delivery semantics. The same option exists for <code class="highlighter-rouge">PersistentActor</code> in akka-persistence by using the <a href="http://doc.akka.io/docs/akka/2.3.8/scala/persistence.html#at-least-once-delivery"><code class="highlighter-rouge">AtLeastOnceDelivery</code></a> trait.</li>
<li>during event processing with at-most-once message delivery semantics. The same option exists for <code class="highlighter-rouge">PersistentActor</code> in akka-persistence by checking whether a consumed event is a live event or a replayed event.</li>
</ul>
<p>Replies from external services are processed like external commands: they may produce new events which may trigger the next step in a distributed business process, for example.</p>
<p>Since EAs can consume events from other EAs, they can also generate any view of application state. An EA can consume events from all other globally or locally distributed producers (EAs) by consuming from the shared, replicated event log. This overcomes a current limitation in akka-persistence where events cannot be easily aggregated from several producers (at least not in a scalable and deterministic way).</p>
<p>If an application wants to restrict an actor to only consume from the event log it should implement the <code class="highlighter-rouge">EventsourcedView</code> (EV) trait (instead of <code class="highlighter-rouge">EventsourcedActor</code>) which implements only the event consumer part of an EA. From a CQRS perspective,</p>
<ul>
<li>EAs should be used to implement the command side (C) of CQRS and maintain a write model (in-memory only in our application)</li>
<li>EVs should be used to implement the query side (Q) of CQRS and maintain a read model (in-memory or persistent in our application)</li>
</ul>
<p>In addition to EAs and EVs, we also plan to implement an interface to <a href="http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-M2/scala.html">akka-streams</a> for producing to and consuming from the distributed event log.</p>
<h2 id="state-replication">State replication</h2>
<p>As already mentioned, by using a replicated event log, application state can be reconstructed (= replicated) on different sites. During inter-site network partitions, sites must remain available for updates to replicated state. Consequently, conflicting updates may occur which must be detected and resolved later when the partition heals. More precisely, conflicting updates may also occur without an inter-site network partition if updates are concurrent.</p>
<p>An example: site A makes an update to the replicated domain object <code class="highlighter-rouge">x1</code> and the corresponding update event <code class="highlighter-rouge">e1</code> is written to the replicated event log. Some times later, site A receives another update event <code class="highlighter-rouge">e2</code> for the same domain object <code class="highlighter-rouge">x1</code> from site B. If site B has processed event <code class="highlighter-rouge">e1</code> before emitting <code class="highlighter-rouge">e2</code>, then <code class="highlighter-rouge">e2</code> causally depends on <code class="highlighter-rouge">e1</code> and site A can simply apply <code class="highlighter-rouge">e2</code> to update <code class="highlighter-rouge">x1</code>. In this case, the two updates, represented by <code class="highlighter-rouge">e1</code> and <code class="highlighter-rouge">e2</code>, have been applied to the replicated domain object <code class="highlighter-rouge">x1</code> on both sites and both copies of <code class="highlighter-rouge">x1</code> converge to the same value. On the other hand, if site B concurrently made an update to <code class="highlighter-rouge">x1</code> (be it because of a network partition or not), there might be a conflict.</p>
<p>Whether concurrent events are also conflicting events completely depends on application logic. For example, concurrent updates to different domain objects may be acceptable to an application whereas concurrent updates to the same domain object may be considered as conflict and must be resolved. Whether any two events are concurrent or have a happened-before relationship (= potential causal relationship) can be determined by comparing their vector timestamps.</p>
<h2 id="conflict-resolution">Conflict resolution</h2>
<p>If application state can be modeled with <a href="http://rbmhtechnology.github.io/eventuate/user-guide.html#commutative-replicated-data-types">commutative replicated data types</a> (CmRDTs) alone, where state update operations are replicated via events, concurrent updates are not an issue at all. However, many state update operations in our application do not commute and we support both interactive and automated conflict resolution strategies.</p>
<p>Conflicting versions of application state are tracked in a concurrent versions tree (where the tree structure is determined by the vector timestamps of contributing events). For any state value of type <code class="highlighter-rouge">S</code> and updates of type <code class="highlighter-rouge">A</code>, concurrent versions of <code class="highlighter-rouge">S</code> can be tracked in a generic way with data type <a href="http://rbmhtechnology.github.io/eventuate/user-guide.html#tracking-conflicting-versions"><code class="highlighter-rouge">ConcurrentVersions</code></a>. Concurrent versions can be tracked for different parts of application state independently, such as individual domain objects or even domain object fields, depending on which granularity level an application wants to detect and resolve conflicts.</p>
<p>During <a href="http://rbmhtechnology.github.io/eventuate/user-guide.html#interactive-conflict-resolution">interactive conflict resolution</a>, a user selects one of the conflicting versions as the “winner”. This selection is stored as explicit conflict resolution event in the event log so that no further user interaction is needed during later event replays. A possible extension could be an interactive merge of conflicting versions. In this case, the conflict resolution event must contain the merge details so that the merge is reproducible.</p>
<p><a href="http://rbmhtechnology.github.io/eventuate/user-guide.html#automated-conflict-resolution">Automated conflict resolution</a> applies a custom conflict resolution function to conflicting versions in order to select a winner. A conflict resolution function could also automatically merge conflicting versions but then we are already in the field of <a href="http://en.wikipedia.org/wiki/Conflict-free_replicated_data_type#State-based_CRDTs">convergent replicated data types</a> (CvRDTs).</p>
Tue, 13 Jan 2015 00:00:00 +0000http://krasserm.github.io/2015/01/13/event-sourcing-at-global-scale/
http://krasserm.github.io/2015/01/13/event-sourcing-at-global-scale/