text:
'use strict';

/**
 * Module dependencies.
 */
var mongoose = require('mongoose'),
	errorHandler = require('./errors.server.controller'),
	Budget = mongoose.model('Budget'),
	_ = require('lodash');

/**
 * Create a budget
 */
exports.create = function(req, res) {
	var budget = new Budget(req.body);
	budget.user = req.user;

	budget.save(function(err) {
		if (err) {
			return res.status(400).send({
				message: errorHandler.getErrorMessage(err)
			});
		} else {
			res.json(budget);
		}
	});
};

/**
 * Show the current budget
 */
exports.read = function(req, res) {
	res.json(req.budget);
};

/**
 * Update a budget
 */
exports.update = function(req, res) {
	var budget = req.budget;

	budget = _.extend(budget, req.body);

	budget.save(function(err) {
		if (err) {
			return res.status(400).send({
				message: errorHandler.getErrorMessage(err)
			});
		} else {
			res.json(budget);
		}
	});
};

/**
 * Delete a budget
 */
exports.delete = function(req, res) {
	var budget = req.budget;

	budget.remove(function(err) {
		if (err) {
			return res.status(400).send({
				message: errorHandler.getErrorMessage(err)
			});
		} else {
			res.json(budget);
		}
	});
};

/**
 * List of Budgets
 */
exports.list = function(req, res) {
	Budget.find().sort('-created').populate('user').exec(function(err, budgets) {
		if (err) {
			return res.status(400).send({
				message: errorHandler.getErrorMessage(err)
			});
		} else {
			res.json(budgets);
		}
	});
};

/**
 * Budget middleware
 */
exports.budgetByID = function(req, res, next, id) {
	Budget.findById(id).populate('user').exec(function(err, budget) {
		if (err) return next(err);
		if (!budget) return next(new Error('Failed to load budget ' + id));
		req.budget = budget;
		next();
	});
};

/**
 * Budget authorization middleware
 */
exports.hasAuthorization = function(req, res, next) {
	if (req.budget.user.id !== req.user.id) {
		return res.status(403).send({
			message: 'User is not authorized'
		});
	}
	next();
};


text:
using Kooboo.IO;
using Kooboo.CMS.Common;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;

namespace Kooboo.CMS.Sites.Extension.ModuleArea.Management
{
    public interface IInstallationFileManager
    {
        #region InstallationContext
        IEnumerable<InstallationContext> AllInstallationLogs(string moduleName);
        void LogInstallation(string moduleName, InstallationContext installationContext);
        InstallationContext GetLastestInstallation(string moduleName);
        #endregion
        IPath GetTempInstallationPath(string moduleName);
        /// <summary>
        /// Save the installation file
        /// </summary>
        /// <param name="moduleStreamEntry">The module stream entry.</param>
        /// <returns></returns>
        string ArchiveTempInstallationPath(string moduleName, string version);
        bool IsInstallationFileExists(InstallationContext installationContext);
        Stream GetInstallationStream(InstallationContext installationContext);
        Stream GetInstallationStream(string moduleName, string installationFile);
        void RemoveHistory(string moduleName);
    }
}

redpajama_set_name: RedPajamaGithub


text:
import zstackwoodpecker.test_state as ts_header
import os
TestAction = ts_header.TestAction
def path():

    return dict(initial_formation="template5", checking_point=8, path_list=[
		[TestAction.create_vm, 'vm1', 'flag=ceph'],
		[TestAction.create_volume, 'volume1', 'flag=ceph,scsi'],
		[TestAction.attach_volume, 'vm1', 'volume1'],
		[TestAction.create_volume, 'volume2', 'flag=ceph,scsi'],
		[TestAction.attach_volume, 'vm1', 'volume2'],
		[TestAction.create_volume, 'volume3', 'flag=ceph,scsi'],
		[TestAction.attach_volume, 'vm1', 'volume3'],
		[TestAction.create_vm_snapshot, 'vm1', 'vm1-snapshot1'],
		[TestAction.create_vm_backup, 'vm1', 'vm1-backup1'],
		[TestAction.stop_vm, 'vm1'],
		[TestAction.use_volume_snapshot, 'volume1-snapshot1'],
		[TestAction.start_vm, 'vm1'],
		[TestAction.create_vm_snapshot, 'vm1', 'vm1-snapshot5'],
		[TestAction.clone_vm, 'vm1', 'vm2'],
		[TestAction.create_vm_snapshot, 'vm1', 'vm1-snapshot9'],
		[TestAction.resize_volume, 'vm2', 5*1024*1024],
		[TestAction.delete_volume_snapshot, 'vm1-snapshot9'],
		[TestAction.stop_vm, 'vm2'],
		[TestAction.reinit_vm, 'vm2'],
		[TestAction.start_vm, 'vm2'],
		[TestAction.delete_vm_snapshot, 'vm1-snapshot1'],
])

'''
The final status:
Running:['vm1', 'vm2']
Stopped:[]
Enadbled:['vm1-snapshot5', 'volume1-snapshot5', 'volume2-snapshot5', 'volume3-snapshot5', 'volume1-snapshot9', 'volume2-snapshot9', 'volume3-snapshot9', 'vm1-backup1', 'volume1-backup1', 'volume2-backup1', 'volume3-backup1']
attached:['volume1', 'volume2', 'volume3']
Detached:[]
Deleted:['vm1-snapshot9', 'vm1-snapshot1', 'volume1-snapshot1', 'volume2-snapshot1', 'volume3-snapshot1']
Expunged:[]
Ha:[]
Group:
	vm_snap2:['vm1-snapshot5', 'volume1-snapshot5', 'volume2-snapshot5', 'volume3-snapshot5']---vm1@volume1_volume2_volume3
	vm_backup1:['vm1-backup1', 'volume1-backup1', 'volume2-backup1', 'volume3-backup1']---vm1@volume1_volume2_volume3
'''


text:
"""Base class for sparse matrices"""

__all__ = ['spmatrix', 'isspmatrix', 'issparse',
        'SparseWarning','SparseEfficiencyWarning']

from warnings import warn

import numpy as np

from sputils import isdense, isscalarlike, isintlike

class SparseWarning(Warning): pass
class SparseFormatWarning(SparseWarning): pass
class SparseEfficiencyWarning(SparseWarning): pass

# The formats that we might potentially understand.
_formats = {'csc':[0, "Compressed Sparse Column"],
            'csr':[1, "Compressed Sparse Row"],
            'dok':[2, "Dictionary Of Keys"],
            'lil':[3, "LInked List"],
            'dod':[4, "Dictionary of Dictionaries"],
            'sss':[5, "Symmetric Sparse Skyline"],
            'coo':[6, "COOrdinate"],
            'lba':[7, "Linpack BAnded"],
            'egd':[8, "Ellpack-itpack Generalized Diagonal"],
            'dia':[9, "DIAgonal"],
            'bsr':[10, "Block Sparse Row"],
            'msr':[11, "Modified compressed Sparse Row"],
            'bsc':[12, "Block Sparse Column"],
            'msc':[13, "Modified compressed Sparse Column"],
            'ssk':[14, "Symmetric SKyline"],
            'nsk':[15, "Nonsymmetric SKyline"],
            'jad':[16, "JAgged Diagonal"],
            'uss':[17, "Unsymmetric Sparse Skyline"],
            'vbr':[18, "Variable Block Row"],
            'und':[19, "Undefined"]
            }

MAXPRINT = 50

class spmatrix(object):
    """ This class provides a base class for all sparse matrices.  It
    cannot be instantiated.  Most of the work is provided by subclasses.
    """

    __array_priority__ = 10.1
    ndim = 2
    def __init__(self, maxprint=MAXPRINT):
        self.format = self.__class__.__name__[:3]
        self._shape = None
        if self.format == 'spm':
            raise ValueError("This class is not intended"
                            " to be instantiated directly.")
        self.maxprint = maxprint

    def set_shape(self,shape):
        shape = tuple(shape)

        if len(shape) != 2:
            raise ValueError("Only two-dimensional sparse arrays "
                                     "are supported.")
        try:
            shape = int(shape[0]),int(shape[1]) #floats, other weirdness
        except:
            raise TypeError('invalid shape')

        if not (shape[0] >= 1 and shape[1] >= 1):
            raise ValueError('invalid shape')

        if (self._shape != shape) and (self._shape is not None):
            try:
                self = self.reshape(shape)
            except NotImplementedError:
                raise NotImplementedError("Reshaping not implemented for %s." %
                                          self.__class__.__name__)
        self._shape = shape

    def get_shape(self):
        return self._shape

    shape = property(fget=get_shape, fset=set_shape)

    def reshape(self,shape):
        raise NotImplementedError

    def astype(self, t):
        return self.tocsr().astype(t).asformat(self.format)

    def asfptype(self):
        """Upcast matrix to a floating point format (if necessary)"""

        fp_types = ['f','d','F','D']

        if self.dtype.char in fp_types:
            return self
        else:
            for fp_type in fp_types:
                if self.dtype <= np.dtype(fp_type):
                    return self.astype(fp_type)

            raise TypeError('cannot upcast [%s] to a floating '
                             'point format' % self.dtype.name)

    def __iter__(self):
        for r in xrange(self.shape[0]):
            yield self[r,:]

    def getmaxprint(self):
        try:
            maxprint = self.maxprint
        except AttributeError:
            maxprint = MAXPRINT
        return maxprint

    #def typecode(self):
    #    try:
    #        typ = self.dtype.char
    #    except AttributeError:
    #        typ = None
    #    return typ

    def getnnz(self):
        try:
            return self.nnz
        except AttributeError:
            raise AttributeError("nnz not defined")

    def getformat(self):
        try:
            format = self.format
        except AttributeError:
            format = 'und'
        return format

    def __repr__(self):
        nnz = self.getnnz()
        format = self.getformat()
        return "<%dx%d sparse matrix of type '%s'\n" \
               "\twith %d stored elements in %s format>" % \
               (self.shape + (self.dtype.type, nnz, _formats[format][1]))

    def __str__(self):
        maxprint = self.getmaxprint()

        A   = self.tocoo()
        nnz = self.getnnz()

        # helper function, outputs "(i,j)  v"
        def tostr(row,col,data):
            triples = zip(zip(row,col),data)
            return '\n'.join( [ ('  %s\t%s' % t) for t in triples] )

        if nnz > maxprint:
            half = maxprint // 2
            out  = tostr(A.row[:half], A.col[:half], A.data[:half])
            out += "\n  :\t:\n"
            half = maxprint - maxprint//2
            out += tostr(A.row[-half:], A.col[-half:], A.data[-half:])
        else:
            out  = tostr(A.row, A.col, A.data)

        return out

    def __nonzero__(self):  # Simple -- other ideas?
        return self.getnnz() > 0

    # What should len(sparse) return? For consistency with dense matrices,
    # perhaps it should be the number of rows?  But for some uses the number of
    # non-zeros is more important.  For now, raise an exception!
    def __len__(self):
        # return self.getnnz()
        raise TypeError("sparse matrix length is ambiguous; use getnnz()"
                         " or shape[0]")

    def asformat(self, format):
        """Return this matrix in a given sparse format

        Parameters
        ----------
        format : {string, None}
            desired sparse matrix format
                - None for no format conversion
                - "csr" for csr_matrix format
                - "csc" for csc_matrix format
                - "lil" for lil_matrix format
                - "dok" for dok_matrix format and so on

        """

        if format is None or format == self.format:
            return self
        else:
            return getattr(self,'to' + format)()

    ###################################################################
    #  NOTE: All arithmetic operations use csr_matrix by default.
    # Therefore a new sparse matrix format just needs to define a
    # .tocsr() method to provide arithmetic support.  Any of these
    # methods can be overridden for efficiency.
    ####################################################################

    def multiply(self, other):
        """Point-wise multiplication by another matrix
        """
        return self.tocsr().multiply(other)

    def dot(self, other):
        return self * other

    def __abs__(self):
        return abs(self.tocsr())

    def __add__(self, other):   # self + other
        return self.tocsr().__add__(other)

    def __radd__(self, other):  # other + self
        return self.tocsr().__radd__(other)

    def __sub__(self, other):   # self - other
        #note: this can't be replaced by self + (-other) for unsigned types
        return self.tocsr().__sub__(other)

    def __rsub__(self, other):  # other - self
        return self.tocsr().__rsub__(other)

    def __mul__(self, other):
        """interpret other and call one of the following

        self._mul_scalar()
        self._mul_vector()
        self._mul_multivector()
        self._mul_sparse_matrix()
        """

        M,N = self.shape

        if isscalarlike(other):
            # scalar value
            return self._mul_scalar(other)

        if issparse(other):
            if self.shape[1] != other.shape[0]:
                raise ValueError('dimension mismatch')
            return self._mul_sparse_matrix(other)

        try:
            other.shape
        except AttributeError:
            # If it's a list or whatever, treat it like a matrix
            other = np.asanyarray(other)

        other = np.asanyarray(other)

        if other.ndim == 1 or other.ndim == 2 and other.shape[1] == 1:
            # dense row or column vector
            if other.shape != (N,) and other.shape != (N,1):
                raise ValueError('dimension mismatch')

            result = self._mul_vector(np.ravel(other))

            if isinstance(other, np.matrix):
                result = np.asmatrix(result)

            if other.ndim == 2 and other.shape[1] == 1:
                # If 'other' was an (nx1) column vector, reshape the result
                result = result.reshape(-1,1)

            return result

        elif other.ndim == 2:
            ##
            # dense 2D array or matrix ("multivector")

            if other.shape[0] != self.shape[1]:
                raise ValueError('dimension mismatch')

            result = self._mul_multivector(np.asarray(other))

            if isinstance(other, np.matrix):
                result = np.asmatrix(result)

            return result
        else:
            raise ValueError('could not interpret dimensions')

    # by default, use CSR for __mul__ handlers
    def _mul_scalar(self, other):
        return self.tocsr()._mul_scalar(other)

    def _mul_vector(self, other):
        return self.tocsr()._mul_vector(other)

    def _mul_multivector(self, other):
        return self.tocsr()._mul_multivector(other)

    def _mul_sparse_matrix(self, other):
        return self.tocsr()._mul_sparse_matrix(other)

    def __rmul__(self, other): # other * self
        if isscalarlike(other):
            return self.__mul__(other)
        else:
            # Don't use asarray unless we have to
            try:
                tr = other.transpose()
            except AttributeError:
                tr = np.asarray(other).transpose()
            return (self.transpose() * tr).transpose()

    ####################
    # Other Arithmetic #
    ####################

    def __truediv__(self, other):
        if isscalarlike(other):
            return self * (1./other)
        else:
            return self.tocsr().__truediv__(other)

    def __div__(self, other):
        # Always do true division
        return self.__truediv__(other)

    def __neg__(self):
        return -self.tocsr()

    def __iadd__(self, other):
        raise NotImplementedError

    def __isub__(self, other):
        raise NotImplementedError

    def __imul__(self, other):
        raise NotImplementedError

    def __idiv__(self, other):
        return self.__itruediv__(other)

    def __itruediv__(self, other):
        raise NotImplementedError

    def __pow__(self, other):
        if self.shape[0] != self.shape[1]:
            raise TypeError('matrix is not square')

        if isintlike(other):
            other = int(other)
            if other < 0:
                raise ValueError('exponent must be >= 0')

            if other == 0:
                from construct import identity
                return identity( self.shape[0], dtype=self.dtype )
            elif other == 1:
                return self.copy()
            else:
                result = self
                for i in range(1,other):
                    result = result*self
                return result
        elif isscalarlike(other):
            raise ValueError('exponent must be an integer')
        else:
            raise NotImplementedError

    def __getattr__(self, attr):
        if attr == 'A':
            return self.toarray()
        elif attr == 'T':
            return self.transpose()
        elif attr == 'H':
            return self.getH()
        elif attr == 'real':
            return self._real()
        elif attr == 'imag':
            return self._imag()
        elif attr == 'size':
            return self.getnnz()
        else:
            raise AttributeError(attr + " not found")

    def transpose(self):
        return self.tocsr().transpose()

    def conj(self):
        return self.tocsr().conj()

    def conjugate(self):
        return self.conj()

    # Renamed conjtranspose() -> getH() for compatibility with dense matrices
    def getH(self):
        return self.transpose().conj()

    def _real(self):
        return self.tocsr()._real()

    def _imag(self):
        return self.tocsr()._imag()

    def nonzero(self):
        """nonzero indices

        Returns a tuple of arrays (row,col) containing the indices
        of the non-zero elements of the matrix.

        Examples
        --------
        >>> from scipy.sparse import csr_matrix
        >>> A = csr_matrix([[1,2,0],[0,0,3],[4,0,5]])
        >>> A.nonzero()
        (array([0, 0, 1, 2, 2]), array([0, 1, 2, 0, 2]))

        """

        # convert to COOrdinate format
        A = self.tocoo()
        nz_mask = A.data != 0
        return (A.row[nz_mask],A.col[nz_mask])

    def getcol(self, j):
        """Returns a copy of column j of the matrix, as an (m x 1) sparse
        matrix (column vector).
        """
        # Spmatrix subclasses should override this method for efficiency.
        # Post-multiply by a (n x 1) column vector 'a' containing all zeros
        # except for a_j = 1
        from csc import csc_matrix
        n = self.shape[1]
        if j < 0:
            j += n
        if j < 0 or j >= n:
            raise IndexError("index out of bounds")
        col_selector = csc_matrix(([1], [[j], [0]]), shape=(n,1), dtype=self.dtype)
        return self * col_selector

    def getrow(self, i):
        """Returns a copy of row i of the matrix, as a (1 x n) sparse
        matrix (row vector).
        """
        # Spmatrix subclasses should override this method for efficiency.
        # Pre-multiply by a (1 x m) row vector 'a' containing all zeros
        # except for a_i = 1
        from csr import csr_matrix
        m = self.shape[0]
        if i < 0:
            i += m
        if i < 0 or i >= m:
            raise IndexError("index out of bounds")
        row_selector = csr_matrix(([1], [[0], [i]]), shape=(1,m), dtype=self.dtype)
        return row_selector * self

    #def __array__(self):
    #    return self.toarray()

    def todense(self):
        return np.asmatrix(self.toarray())

    def toarray(self):
        return self.tocoo().toarray()

    def todok(self):
        return self.tocoo().todok()

    def tocoo(self):
        return self.tocsr().tocoo()

    def tolil(self):
        return self.tocsr().tolil()

    def todia(self):
        return self.tocoo().todia()

    def tobsr(self, blocksize=None):
        return self.tocsr().tobsr(blocksize=blocksize)

    def copy(self):
        return self.__class__(self,copy=True)

    def sum(self, axis=None):
        """Sum the matrix over the given axis.  If the axis is None, sum
        over both rows and columns, returning a scalar.
        """
        # We use multiplication by an array of ones to achieve this.
        # For some sparse matrix formats more efficient methods are
        # possible -- these should override this function.
        m, n = self.shape
        if axis == 0:
            # sum over columns
            return np.asmatrix(np.ones((1, m), dtype=self.dtype)) * self
        elif axis == 1:
            # sum over rows
            return self * np.asmatrix(np.ones((n, 1), dtype=self.dtype))
        elif axis is None:
            # sum over rows and columns
            return ( self * np.asmatrix(np.ones((n, 1), dtype=self.dtype)) ).sum()
        else:
            raise ValueError("axis out of bounds")

    def mean(self, axis=None):
        """Average the matrix over the given axis.  If the axis is None,
        average over both rows and columns, returning a scalar.
        """
        if axis == 0:
            mean = self.sum(0)
            mean *= 1.0 / self.shape[0]
            return mean
        elif axis == 1:
            mean = self.sum(1)
            mean *= 1.0 / self.shape[1]
            return mean
        elif axis is None:
            return self.sum(None) * 1.0 / (self.shape[0]*self.shape[1])
        else:
            raise ValueError("axis out of bounds")

    def diagonal(self):
        """Returns the main diagonal of the matrix
        """
        #TODO support k != 0
        return self.tocsr().diagonal()

    def setdiag(self, values, k=0):
        """Fills the diagonal elements {a_ii} with the values from the
        given sequence.  If k != 0, fills the off-diagonal elements
        {a_{i,i+k}} instead.

        values may have any length.  If the diagonal is longer than values,
        then the remaining diagonal entries will not be set.  If values if
        longer than the diagonal, then the remaining values are ignored.
        """
        M, N = self.shape
        if (k > 0 and k >= N) or (k < 0 and -k >= M):
            raise ValueError("k exceedes matrix dimensions")
        if k < 0:
            max_index = min(M+k, N, len(values))
            for i,v in enumerate(values[:max_index]):
                self[i - k, i] = v
        else:
            max_index = min(M, N-k, len(values))
            for i,v in enumerate(values[:max_index]):
                self[i, i + k] = v

from sputils import _isinstance

def isspmatrix(x):
    return _isinstance(x, spmatrix)

issparse = isspmatrix


text:
/**
 * Bootable
 * @mixin
 *
 * Used to add lazy content functionality to components
 * Looks for change in "isActive" to automatically boot
 * Otherwise can be set manually
 */
export default {
  data: () => ({
    isBooted: false
  }),

  props: {
    lazy: Boolean
  },

  watch: {
    isActive () {
      this.isBooted = true
    }
  },

  methods: {
    showLazyContent (content) {
      return (this.isBooted || !this.lazy)
        ? content
        : null
    }
  }
}


text:
import {Route, Segment} from './types'

const VALID_PARAM_SEGMENT = /^[a-zA-Z0-9_-]+$/

function createSegment(segment: string): Segment | null {
  if (!segment) {
    return null
  }
  if (segment.startsWith(':')) {
    const paramName = segment.substring(1)
    if (!VALID_PARAM_SEGMENT.test(paramName)) {
      const addendum = segment.includes('*')
        ? ' Splats are not supported. Consider using child routes instead'
        : ''
      // eslint-disable-next-line no-console
      console.error(
        new Error(`Warning: Param segments "${segment}" includes invalid characters.${addendum}`)
      )
    }
    return {type: 'param', name: paramName}
  }
  return {type: 'dir', name: segment}
}

export default function parseRoute(route: string): Route {
  const [pathname] = route.split('?')

  const segments = pathname.split('/').map(createSegment).filter(Boolean) as Segment[]

  return {
    raw: route,
    segments: segments,
  }
}


text:
.. _str: https://docs.python.org/2/library/stdtypes.html

.. _vim.TaskReason: ../vim/TaskReason.rst

vim.TaskReasonUser
==================
  Indicates that the task was queued by a specific user.
:extends: vim.TaskReason_

Attributes:
    userName (`str`_):

       Name of the user that queued the task.


text:
/* global $, TextSummary, _, hljs, TWITTER_USER */
'use strict';

var markdown = function(s) {
  return window.markdownit().render(s);
};

var OUTPUT_LANG = 'en';

var globalState = {
  twitterUserId: undefined,
  selectedTwitterUser: undefined,
  selectedTwitterImage: undefined,
  selectedTwitterUserLang: undefined,
  selectedSample: undefined,
  languageSelected: undefined,
  currentProfile: undefined,
  userLocale: undefined
};

var QUERY_PARAMS = (function(a) {
  if (a == '')
    return {};
  var b = {};
  for (var i = 0; i < a.length; ++i) {
    var p = a[i].split('=', 2);
    if (p.length == 1)
      b[p[0]] = '';
    else
      b[p[0]] = decodeURIComponent(p[1].replace(/\+/g, ' '));
  }
  return b;
})(window.location.search.substr(1).split('&'));

function getBrowserLang() {
  if (navigator.languages != undefined)
    return navigator.languages[0];
  else
    return navigator.language;
}

function getBrowserLangNoLocale() {
  var lang = getBrowserLang();
  return lang.substring(0, 2);
}

function extend(target, source) {
  Object.keys(source).forEach(function(k) {
    target[k] = source[k];
  });
  return target;
}

function clone(o) {
  return extend({}, o);
}

function replaces(s, replaces) {
  var out = s;
  replaces.forEach(function(r) {
    out = out.replace(r.search, r.replace);
  });

  return out;
}

function renderMarkdown(s) {
  return replaces(markdown(s || ''), [
    {
      search: /\<\a /g,
      replace: '<a class="base--a" target="_blank" '
    }
  ]);
}

$(document).ready(function() {

  var SAMPLE_TEXTS = [ 'sample1', 'sample2', 'sample3', 'ar', 'ja'];
  var textCache = {};

  globalState.selectedSample = SAMPLE_TEXTS[0];
  globalState.languageSelected = undefined;

  var $big5Traits = $('.output-big-5--traits');
  var $needsTraits = $('.output-needs--traits');
  var $needsMoreTraits = $('.output-needs--more-traits');
  var $valuesTraits = $('.output-values--traits');
  var $needsToggle = $('.output-needs--toggle');
  var $outputSummaryText = $('.output-summary--summary');
  var $inputTextArea = $('.input--text-area');
  var $inputWordCount = $('.input--word-count-number');
  var $inputForm1 = $('.input--form1');
  var $inputForm2 = $('.input--form2');
  var $resetButton = $('.input--reset-button');
  var $loading = $('.loading');
  var $output = $('.output');
  var $outputHeader = $('.output--header');
  var $outputJSON = $('.output--json');
  var $outputJSONCode = $('.output--json-code');
  var $outputJSONButton = $('.output--json-button');
  var $error = $('.error');
  var $errorMessage = $('.error--message');

  // Instantiate external PI modules
  const TraitNames = new PersonalityTraitNames({
    version : 'v3',
    locale : globalState.userLocale || OUTPUT_LANG
  });

  const TraitDescriptions = new PersonalityTraitDescriptions({
    version: 'v3',
    locale: globalState.userLocale || OUTPUT_LANG,
    format: 'markdown'
  });

  const ConsumptionPreferences = new PersonalityConsumptionPreferences({
    version: 'v3',
    locale: globalState.userLocale || OUTPUT_LANG
  });

  function setTextSample(value, readonly) {
    $('#inputText').val(value);
    if (readonly) {
      $('#inputText').attr('readonly', 'readonly');
    } else {
      $('#inputText').removeAttr('readonly');
    }
  }

  function setLoadingState() {
    resetOutputs();
    $loading.show();
    scrollTo($loading);
  }

  function loadTwitterUser(twitterHandle, options) {
    setLoadingState();
    getProfileForTwitterUser(twitterHandle, options);
  }

  function registerHandlers() {
    globalState.userLocale = getBrowserLangNoLocale();

    $('input[name="text-lang"]').click(function() {
      globalState.selectedLanguage = $(this).attr('value');
    });

    $('input[name="text-sample"]').click(function() {
      var textFile = $(this).attr('data-file'),
        orientation = $(this).attr('data-orientation');
      globalState.selectedSample = textFile;

      if (orientation === 'right-to-left') {
        $inputTextArea.removeClass('left-to-right');
        $inputTextArea.addClass('right-to-left');
      } else {
        $inputTextArea.removeClass('right-to-left');
        $inputTextArea.addClass('left-to-right');
      }

      $('#languageChooser').hide();

      loadSampleText(textFile);
      updateWordCount();
    });

    $(window).resize(function() {
      if ($(window).width() < 800) {
        $('.smartphone-hidden').hide();
        if (globalState.selectedSample == 'custom') {
          $('input[name="text-sample"]:first').trigger('click');
        }
      } else {
        $('label[for="text-custom"]').show();
      }
    });

    $('input#text-custom').unbind('click').click(function() {
      globalState.selectedSample = 'custom';

      $inputTextArea.removeClass('right-to-left');
      $inputTextArea.addClass('left-to-right');

      $('#languageChooser').show();
      setTextSample('', false);
      updateWordCount();

      if (!globalState.selectedLanguage) {
        $('input#lang-en').trigger('click');
      }
    });

    $('input[name="twitter"]').click(function() {
      var twitterId = $(this).val();
      var twitterLang = $(this).attr('data-lang');
      globalState.selectedTwitterUser = twitterId;
      globalState.selectedTwitterImage = $('label[for="' + $(this).attr('id') + '"] img').attr('src');
      globalState.selectedTwitterUserLang = twitterLang;
    });

    $inputForm1.submit(function(e) {
      e.cancelBubble = true;
      e.preventDefault();
      if (e.stopPropagation)
        e.stopPropagation();

      enableAnalyzeButtons(false);
      loadTwitterUser(globalState.selectedTwitterUser, {
        language: globalState.selectedTwitterUserLang
      });
    });

    $inputForm2.submit(function(e) {
      e.cancelBubble = true;
      e.preventDefault();
      if (e.stopPropagation)
        e.stopPropagation();

      enableAnalyzeButtons(false);
      var lang = globalState.selectedSample == 'custom'
        ? globalState.selectedLanguage
        : $('input#text-' + globalState.selectedSample).attr('data-lang');

      setLoadingState();

      getProfileForText($('.input--text-area').val(), {language: lang});
    });
  }

  function setTextSummary(profile) {
    var textSummary = new TextSummary({ version: 'v3', locale: globalState.userLocale || OUTPUT_LANG});
    var summary = textSummary.getSummary(profile);
    $('#personalitySummary').empty();
    $('#personalitySummary').append('
' + summary.split('\n').join('

') + '
');
  }

  /**
   * Toggle Big 5 Subtraits
   */
  $(document).on('click', '.output-big-5--trait-label', function() {
    $(this).closest('.percent-bar-and-score').toggleClass('toggled');
  });

  $resetButton.click(function() {
    $('input[name="twitter"]:first').trigger('click');
    $('input[name="text-sample"]:first').trigger('click');
    $('.tab-panels--tab:first').trigger('click');
    $('#your-twitter-panel .auth-form').show();
    $('#your-twitter-panel .analysis-form').hide();
    resetOutputs();
    selectDefaultLanguage();
  });

  // toggleNeedsTraits
  $needsToggle.click(function() {
    $needsMoreTraits.toggle();
    $needsToggle.text($needsToggle.text() == '<<' ? '>>' : '<<');
  });

  $outputJSONButton.click(function() {
    $outputJSON.toggle();
    scrollTo($outputJSON);
  });

  function getProfileForTwitterUser(userId, options) {
    getProfile(userId, extend(options || {}, { source_type: 'twitter' }));
  }

  function getProfileForText(text, options) {
    getProfile(text, extend(options || {}, {source_type: 'text'}));
  }

  /**
  * Localization
  */
  function replacementsForLang(lang) {
    var replacements = {
      'en': {
        'Extraversion': 'Introversion/Extraversion',
        'Outgoing': 'Warmth',
        'Uncompromising': 'Straightforwardness',
        'Immoderation': 'Impulsiveness',
        'Susceptible to stress': 'Sensitivity to stress',
        'Conservation': 'Tradition',
        'Openness to change': 'Stimulation',
        'Hedonism': 'Taking pleasure in life',
        'Self-enhancement': 'Achievement',
        'Self-transcendence': 'Helping others'
      },
      'ja': {
        'Openness': '知的好奇心',
        'Friendliness': '友好性'
      }
    };

    return replacements[lang] || {};
  }

  function getErrorMessage(error) {
    var message = GENERIC_REQUEST_ERROR;
    if (error.responseJSON && error.responseJSON.error) {
      message = error.responseJSON.error.error;
    } else if (error.responseJSON && error.responseJSON.message) {
      message = error.responseJSON.message
    }
    return message;
  }

  function defaultProfileOptions(options) {
    var defaults = extend({
      source_type: 'text',
      accept_language: globalState.userLocale || OUTPUT_LANG,
      include_raw: false,
      consumption_preferences: true
    }, options || {});

    if (defaults.source_type !== 'twitter') {
      defaults = extend({
        language: globalState.userLocale || OUTPUT_LANG
      }, defaults);
    }
    return defaults;
  }

  function enableAnalyzeButtons(value) {
    $('.input--submit-button1').prop('disabled', !value);
    $('.input--submit-button2').prop('disabled', !value);
  }

  function getProfile(data, options) {
    options = defaultProfileOptions(options);

    var payload = clone(options),
      url = '/api/profile/' + options.source_type;

    if (options.source_type === 'twitter')
      payload.userId = data;
    else
      payload.text = data;

    $.ajax({
      type: 'POST',
      data: payload,
      url: url,
      dataType: 'json',
      success: function(data) {
        $loading.hide();
        $output.show();
        scrollTo($outputHeader);
        loadOutput(data);
        updateJSON(data);
        loadConsumptionPreferences(data);
        enableAnalyzeButtons(true);

      },
      error: function(err) {
        // eslint-disable-next-line
        console.error(err);
        $loading.hide();
        $error.show();
        $errorMessage.text(getErrorMessage(err));
        enableAnalyzeButtons(true);
      }
    });
  }

  /**
  * cpIdMapping returns the description for a consumption_preference_id
  * Uses the personality-consumption-preferences npm module
  */
  function cpIdMapping(consumption_preference_id) {
    return ConsumptionPreferences.description(consumption_preference_id);
  }

  function cpIdSortingLikely(cpid, lang) {
     var sortArray;
     if(lang == 'en') sortArray = enSortLikely;
     if(lang == 'es') sortArray = esSortLikely;
     if(lang == 'ja') sortArray = jaSortLikely;
     if(lang == 'ar') sortArray = arSortLikely;
     if(lang == 'ko') sortArray = koSortLikely;
     return sortArray.indexOf(cpid);
  }

  function cpIdSortingUnlikely(cpid, lang) {
     var sortArray;
     if(lang == 'en') sortArray = enSortUnlikely;
     if(lang == 'es') sortArray = esSortUnlikely;
     if(lang == 'ja') sortArray = jaSortUnlikely;
     if(lang == 'ar') sortArray = arSortUnlikely;
     if(lang == 'ko') sortArray = koSortUnlikely;
     return sortArray.indexOf(cpid);
  }

  var consumptionPrefMusic = new Set([
    'consumption_preferences_music_rap',
    'consumption_preferences_music_country',
    'consumption_preferences_music_r_b',
    'consumption_preferences_music_hip_hop',
    'consumption_preferences_music_live_event',
    'consumption_preferences_music_playing',
    'consumption_preferences_music_latin',
    'consumption_preferences_music_rock',
    'consumption_preferences_music_classical'
  ]);

  var consumptionPrefMovie = new Set([
    'consumption_preferences_movie_romance',
    'consumption_preferences_movie_adventure',
    'consumption_preferences_movie_horror',
    'consumption_preferences_movie_musical',
    'consumption_preferences_movie_historical',
    'consumption_preferences_movie_science_fiction',
    'consumption_preferences_movie_war',
    'consumption_preferences_movie_drama',
    'consumption_preferences_movie_action',
    'consumption_preferences_movie_documentary'
  ]);

  function addIfAllowedReducer(accumulator, toadd) {
    if (consumptionPrefMusic.has(toadd.cpid)) {
      if (accumulator.reduce(function(k, v) {
        return consumptionPrefMusic.has(v.cpid)
          ? k + 1
          : k;
      }, 0) < 1) {
        accumulator.push(toadd);
      }
    } else if (consumptionPrefMovie.has(toadd.cpid)) {

      if (accumulator.reduce(function(k, v) {
        return consumptionPrefMovie.has(v.cpid)
          ? k + 1
          : k;
      }, 0) < 1) {
        accumulator.push(toadd);
      }
    } else {
      accumulator.push(toadd);
    }
    return accumulator;
  }

  function sortIdxComparator(x, y) {

    var a = x.idx;
    var b = y.idx;

    if (a  b) {
      return 1;
    }

    if (a === b) {
      return 0;
    }
  }

  function loadConsumptionPreferences(data) {
    var cpsect = $('.output-summary--consumption-behaviors--section');
    var behaviors = $('.output-summary--consumption-behaviors--section');
    var behaviors_likely = $('.output-summary--likely-behaviors');
    var behaviors_unlikely = $('.output-summary--unlikely-behaviors');
    var lang = data.processed_language;

    if (data.consumption_preferences) {
      var likelycps = data.consumption_preferences.reduce(function(k, v) {
        v.consumption_preferences.map(function(child_item) {
          if (child_item.score === 1) {
            k.push({
              name: cpIdMapping(child_item.consumption_preference_id),
              idx: cpIdSortingLikely(child_item.consumption_preference_id,lang),
              cpid: child_item.consumption_preference_id
            });
          }
        });
        return k;
      }, []);

      var unlikelycps = data.consumption_preferences.reduce(function(k, v) {
        v.consumption_preferences.map(function(child_item) {
          if (child_item.score === 0) {
            k.push({
              name: cpIdMapping(child_item.consumption_preference_id),
              idx: cpIdSortingUnlikely(child_item.consumption_preference_id,lang),
              cpid: child_item.consumption_preference_id
            });
          }
        });
        return k;
      },[]);

      behaviors_likely.empty();
      likelycps.sort(sortIdxComparator).reduce(addIfAllowedReducer, []).slice(0, 3).map(function(item) {
        behaviors_likely.append("
" + item.name + "
\n");
      });

      behaviors_unlikely.empty();
      unlikelycps.sort(sortIdxComparator).reduce(addIfAllowedReducer, []).slice(0, 3).map(function(item) {
        behaviors_unlikely.append('
' + item.name + '
\n');
      });

      behaviors_likely.show();
      behaviors_unlikely.show();
    } else {
      behaviors_likely.hide();
      behaviors_unlikely.hide();
    }
  }

  const replacements = replacementsForLang(globalState.userLocale || OUTPUT_LANG);

  function loadOutput(data) {
    setTextSummary(data);
    loadWordCount(data);

    // Add wrapped traits data from the user profile into the html
    $big5Traits.append(_.template(big5PercentTemplate.innerHTML, {
      items: wrapTraits(data).sort(sortScores),
      tooltips: function(traitId) {
        return renderMarkdown(TraitDescriptions.description(traitId));
      }
    }));

    // Add wrapped needs data from the specified user profile into the html
    $needsTraits.append(_.template(outputStatsPercentTemplate.innerHTML, {
      items: wrapNeeds(data).sort(sortScores).slice(0, 5),
      tooltips: function(traitId) {
        return renderMarkdown(TraitDescriptions.description(traitId));
      }
    }));

    // Add wrapped needs 'more' data from the specified user profile into the html
    $needsMoreTraits.append(_.template(outputStatsPercentTemplate.innerHTML, {
      items: wrapNeeds(data).sort(sortScores).slice(5, wrapNeeds(data).length),
      tooltips: function(traitId) {
        return renderMarkdown(TraitDescriptions.description(traitId));
      }
    }));

    // Add wrapped values data from the specified user profile into the html
    $valuesTraits.append(_.template(outputStatsPercentTemplate.innerHTML, {
      items: wrapValues(data).sort(sortScores),
      tooltips: function(traitId) {
        return renderMarkdown(TraitDescriptions.description(traitId));
      }
    }));

    // NOTE: v3 update - is this necessary here? - should it be moved elsewhere?
    globalState.currentProfile = data;

  }

  function wrapTraits(data){
    return data.personality.map(function(obj) {
      const traitName = TraitNames.name(obj.trait_id);
      return {
        name: replacements[traitName] ? replacements[traitName] : traitName,
        id: obj.trait_id,
        score: Math.round(obj.percentile * 100),
        children: obj.children.map(function(obj2) {
          const traitName2 = TraitNames.name(obj2.trait_id);
          return {
            name: replacements[traitName2] ? replacements[traitName2] : traitName2,
            id: obj2.trait_id,
            score: Math.round(obj2.percentile * 100)
          }
        }).sort(function(a, b) { return b.score - a.score; })
      }
    });
  }

  function wrapNeeds(data) {
    return data.needs.map(function(obj) {
      const traitName = TraitNames.name(obj.trait_id);
      return {
        id: obj.trait_id,
        name: replacements[traitName] ? replacements[traitName] : traitName,
        score: Math.round(obj.percentile * 100)
      }
    });
  }

  function wrapValues(data) {
    return data.values.map(function(obj) {
      const traitName = TraitNames.name(obj.trait_id);
      return {
        id: obj.trait_id,
        name: replacements[traitName] ? replacements[traitName] : traitName,
        score: Math.round(obj.percentile * 100)
      };
    });
  }

  $inputTextArea.on('propertychange change click keyup input paste', function() {
    updateWordCount();
  });

  function loadWordCount(data) {
    $('.output--word-count-number').text(data.word_count);
    $('.output--word-count-message').removeClass('show');
    if (data.processed_lang === 'en') {
      if (data.word_count >= 3000)
        $('.output--word-count-message_VERY-STRONG_NEW_MODEL').addClass('show');
      else if (data.word_count < 3000 && data.word_count >= 1200)
        $('.output--word-count-message_STRONG_NEW_MODEL').addClass('show');
      else if (data.word_count < 1200 && data.word_count >= 600)
        $('.output--word-count-message_DECENT_NEW_MODEL').addClass('show');
      else
        $('.output--word-count-message_WEAK_NEW_MODEL').addClass('show');
    }
    else {
      if (data.word_count > 6000)
        $('.output--word-count-message_VERY-STRONG').addClass('show');
      else if (data.word_count <= 6000 && data.word_count >= 3500)
        $('.output--word-count-message_STRONG').addClass('show');
      else if (data.word_count < 3500 && data.word_count >= 1500)
        $('.output--word-count-message_DECENT').addClass('show');
      else
        $('.output--word-count-message_WEAK').addClass('show');
    }
  }

  function scrollTo(element) {
    $('html, body').animate({
      scrollTop: element.offset().top
    }, 'fast');
  }

  function resetOutputs() {
    $output.hide();
    $error.hide();
    $loading.hide();
    $big5Traits.empty();
    $needsTraits.empty();
    $needsMoreTraits.empty();
    $valuesTraits.empty();
    $('.output-big-5--sub-tree').hide();
    $needsMoreTraits.hide();
    $outputSummaryText.empty();
    $outputJSONCode.empty();
    $outputJSON.hide();
  }

  function sortScores(obj1, obj2) {
    return obj2.score - obj1.score;
  }

  function preloadSampleTexts(callback) {
    var shared = {
      done: 0
    };
    SAMPLE_TEXTS.forEach(function(name) {
      $Q.get('data/text/' + name + '.txt').then(function(text) {
        shared.done = shared.done + 1;
        textCache[name] = text;

        if (shared.done == SAMPLE_TEXTS.length && callback) {
          callback();
        }
      }).done();
    });
  }

  function loadSampleText(name) {
    if (textCache[name]) {
      setTextSample(textCache[name], true);
      updateWordCount();
    } else {
      $Q.get('data/text/' + name + '.txt').then(function(text) {
        setTextSample(text, true);
        textCache[name] = text;
      }).then(function() {
        updateWordCount();
      }).done();
    }
  }

  function showHiddenLanguages() {
    var enableLang = {
      'ar': function() {
        $('label[for="text-ar"]').show();
        $('label[for="lang-ar"]').show();
      }
    };

    Object.keys($.url().param()).filter(function(p) {
      return p.slice(0, 5) === 'lang-';
    }).map(function(p) {
      return p.slice(5, p.length);
    }).forEach(function(lang) {
      if (enableLang[lang]) {
        enableLang[lang]();
      }
    });
  }

  function selfAnalysis() {
    return QUERY_PARAMS.source == 'myself';
  }

  function setSelfAnalysis() {
    console.log('Analyzing twitter user ', '@' + TWITTER_USER);
    globalState.twitterUserId = TWITTER_USER.handle;
    globalState.twitterUserImage = TWITTER_USER.image;
    loadTwitterUser(TWITTER_USER.handle, {live_crawling: true});
    $('#self-analysis-tab').trigger('click');
    $('#your-twitter-panel .auth-form').hide();
    $('#your-twitter-panel .analysis-form label').remove();
    $('#your-twitter-panel .analysis-form').append([
      '', '@',
      TWITTER_USER.handle,
      ''
    ].join(''));
    $('#my-twitter').trigger('click');
    $('#your-twitter-panel .analysis-form').show();
  }

  function initialize() {
    $('input[name="twitter"]:first').attr('checked', true);
    $('input[name="text-sample"]:first').attr('checked', true);

    globalState.selectedTwitterUser = $('input[name="twitter"]:first').val();
    showHiddenLanguages();
    preloadSampleTexts(function() {
      loadSampleText(globalState.selectedSample);
    });
    registerHandlers();
    $inputTextArea.addClass('orientation', 'left-to-right');

    if (selfAnalysis() && TWITTER_USER.handle) {
      setSelfAnalysis();
    }
    selectDefaultLanguage();
  }

  function selectDefaultLanguage() {
    if (['en', 'es', 'ja', 'ar'].indexOf(globalState.userLocale) >= 0) {
      $('#lang-' + globalState.userLocale).prop('checked', true).trigger('click');
    }
  }

  function countWords(str) {
    return str.split(' ').length;
  }

  function updateWordCount() {
    $inputWordCount.text(countWords($inputTextArea.val()));
  }

  function updateJSON(results) {
    $outputJSONCode.html(JSON.stringify(results, null, 2));
    $('.code--json').each(function(i, b) {
      hljs.highlightBlock(b);
    });
  }

  initialize();
});


text:
package org.apache.spark.sql.execution.streaming.state

import java.io._
import java.util
import java.util.Locale
import java.util.concurrent.atomic.LongAdder

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.util.control.NonFatal

import com.google.common.io.ByteStreams
import org.apache.commons.io.IOUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._

import org.apache.spark.{SparkConf, SparkEnv}
import org.apache.spark.internal.Logging
import org.apache.spark.io.CompressionCodec
import org.apache.spark.sql.catalyst.expressions.UnsafeRow
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.streaming.CheckpointFileManager
import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.{SizeEstimator, Utils}

/**
 * An implementation of [[StateStoreProvider]] and [[StateStore]] in which all the data is backed
 * by files in an HDFS-compatible file system. All updates to the store has to be done in sets
 * transactionally, and each set of updates increments the store's version. These versions can
 * be used to re-execute the updates (by retries in RDD operations) on the correct version of
 * the store, and regenerate the store version.
 *
 * Usage:
 * To update the data in the state store, the following order of operations are needed.
 *
 * // get the right store
 * - val store = StateStore.get(
 *      StateStoreId(checkpointLocation, operatorId, partitionId), ..., version, ...)
 * - store.put(...)
 * - store.remove(...)
 * - store.commit()    // commits all the updates to made; the new version will be returned
 * - store.iterator()  // key-value data after last commit as an iterator
 *
 * Fault-tolerance model:
 * - Every set of updates is written to a delta file before committing.
 * - The state store is responsible for managing, collapsing and cleaning up of delta files.
 * - Multiple attempts to commit the same version of updates may overwrite each other.
 *   Consistency guarantees depend on whether multiple attempts have the same updates and
 *   the overwrite semantics of underlying file system.
 * - Background maintenance of files ensures that last versions of the store is always recoverable
 * to ensure re-executed RDD operations re-apply updates on the correct past version of the
 * store.
 */
private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with Logging {

  class HDFSBackedReadStateStore(val version: Long, map: HDFSBackedStateStoreMap)
    extends ReadStateStore {

    override def id: StateStoreId = HDFSBackedStateStoreProvider.this.stateStoreId

    override def get(key: UnsafeRow): UnsafeRow = map.get(key)

    override def iterator(): Iterator[UnsafeRowPair] = {
      map.iterator()
    }

    override def abort(): Unit = {}

    override def toString(): String = {
      s"HDFSReadStateStore[id=(op=${id.operatorId},part=${id.partitionId}),dir=$baseDir]"
    }

    override def prefixScan(prefixKey: UnsafeRow): Iterator[UnsafeRowPair] = {
      map.prefixScan(prefixKey)
    }
  }

  /** Implementation of [[StateStore]] API which is backed by an HDFS-compatible file system */
  class HDFSBackedStateStore(val version: Long, mapToUpdate: HDFSBackedStateStoreMap)
    extends StateStore {

    /** Trait and classes representing the internal state of the store */
    trait STATE
    case object UPDATING extends STATE
    case object COMMITTED extends STATE
    case object ABORTED extends STATE

    private val newVersion = version + 1
    @volatile private var state: STATE = UPDATING
    private val finalDeltaFile: Path = deltaFile(newVersion)
    private lazy val deltaFileStream = fm.createAtomic(finalDeltaFile, overwriteIfPossible = true)
    private lazy val compressedStream = compressStream(deltaFileStream)

    override def id: StateStoreId = HDFSBackedStateStoreProvider.this.stateStoreId

    override def get(key: UnsafeRow): UnsafeRow = {
      mapToUpdate.get(key)
    }

    override def put(key: UnsafeRow, value: UnsafeRow): Unit = {
      require(value != null, "Cannot put a null value")
      verify(state == UPDATING, "Cannot put after already committed or aborted")
      val keyCopy = key.copy()
      val valueCopy = value.copy()
      mapToUpdate.put(keyCopy, valueCopy)
      writeUpdateToDeltaFile(compressedStream, keyCopy, valueCopy)
    }

    override def remove(key: UnsafeRow): Unit = {
      verify(state == UPDATING, "Cannot remove after already committed or aborted")
      val prevValue = mapToUpdate.remove(key)
      if (prevValue != null) {
        writeRemoveToDeltaFile(compressedStream, key)
      }
    }

    /** Commit all the updates that have been made to the store, and return the new version. */
    override def commit(): Long = {
      verify(state == UPDATING, "Cannot commit after already committed or aborted")

      try {
        commitUpdates(newVersion, mapToUpdate, compressedStream)
        state = COMMITTED
        logInfo(s"Committed version $newVersion for $this to file $finalDeltaFile")
        newVersion
      } catch {
        case NonFatal(e) =>
          throw new IllegalStateException(
            s"Error committing version $newVersion into $this", e)
      }
    }

    /** Abort all the updates made on this store. This store will not be usable any more. */
    override def abort(): Unit = {
      // This if statement is to ensure that files are deleted only once: if either commit or abort
      // is called before, it will be no-op.
      if (state == UPDATING) {
        state = ABORTED
        cancelDeltaFile(compressedStream, deltaFileStream)
      } else {
        state = ABORTED
      }
      logInfo(s"Aborted version $newVersion for $this")
    }

    /**
     * Get an iterator of all the store data.
     * This can be called only after committing all the updates made in the current thread.
     */
    override def iterator(): Iterator[UnsafeRowPair] = mapToUpdate.iterator()

    override def prefixScan(prefixKey: UnsafeRow): Iterator[UnsafeRowPair] = {
      mapToUpdate.prefixScan(prefixKey)
    }

    override def metrics: StateStoreMetrics = {
      // NOTE: we provide estimation of cache size as "memoryUsedBytes", and size of state for
      // current version as "stateOnCurrentVersionSizeBytes"
      val metricsFromProvider: Map[String, Long] = getMetricsForProvider()

      val customMetrics = metricsFromProvider.flatMap { case (name, value) =>
        // just allow searching from list cause the list is small enough
        supportedCustomMetrics.find(_.name == name).map(_ -> value)
      } + (metricStateOnCurrentVersionSizeBytes -> SizeEstimator.estimate(mapToUpdate))

      StateStoreMetrics(mapToUpdate.size(), metricsFromProvider("memoryUsedBytes"), customMetrics)
    }

    /**
     * Whether all updates have been committed
     */
    override def hasCommitted: Boolean = {
      state == COMMITTED
    }

    override def toString(): String = {
      s"HDFSStateStore[id=(op=${id.operatorId},part=${id.partitionId}),dir=$baseDir]"
    }
  }

  def getMetricsForProvider(): Map[String, Long] = synchronized {
    Map("memoryUsedBytes" -> SizeEstimator.estimate(loadedMaps),
      metricLoadedMapCacheHit.name -> loadedMapCacheHitCount.sum(),
      metricLoadedMapCacheMiss.name -> loadedMapCacheMissCount.sum())
  }

  /** Get the state store for making updates to create a new `version` of the store. */
  override def getStore(version: Long): StateStore = {
    val newMap = getLoadedMapForStore(version)
    logInfo(s"Retrieved version $version of ${HDFSBackedStateStoreProvider.this} for update")
    new HDFSBackedStateStore(version, newMap)
  }

  /** Get the state store for reading to specific `version` of the store. */
  override def getReadStore(version: Long): ReadStateStore = {
    val newMap = getLoadedMapForStore(version)
    logInfo(s"Retrieved version $version of ${HDFSBackedStateStoreProvider.this} for readonly")
    new HDFSBackedReadStateStore(version, newMap)
  }

  private def getLoadedMapForStore(version: Long): HDFSBackedStateStoreMap = synchronized {
    require(version >= 0, "Version cannot be less than 0")
    val newMap = HDFSBackedStateStoreMap.create(keySchema, numColsPrefixKey)
    if (version > 0) {
      newMap.putAll(loadMap(version))
    }
    newMap
  }

  override def init(
      stateStoreId: StateStoreId,
      keySchema: StructType,
      valueSchema: StructType,
      numColsPrefixKey: Int,
      storeConf: StateStoreConf,
      hadoopConf: Configuration): Unit = {
    this.stateStoreId_ = stateStoreId
    this.keySchema = keySchema
    this.valueSchema = valueSchema
    this.storeConf = storeConf
    this.hadoopConf = hadoopConf
    this.numberOfVersionsToRetainInMemory = storeConf.maxVersionsToRetainInMemory

    require((keySchema.length == 0 && numColsPrefixKey == 0) ||
      (keySchema.length > numColsPrefixKey), "The number of columns in the key must be " +
      "greater than the number of columns for prefix key!")
    this.numColsPrefixKey = numColsPrefixKey

    fm.mkdirs(baseDir)
  }

  override def stateStoreId: StateStoreId = stateStoreId_

  /** Do maintenance backing data files, including creating snapshots and cleaning up old files */
  override def doMaintenance(): Unit = {
    try {
      doSnapshot()
      cleanup()
    } catch {
      case NonFatal(e) =>
        logWarning(s"Error performing snapshot and cleaning up $this")
    }
  }

  override def close(): Unit = {
    loadedMaps.values.asScala.foreach(_.clear())
  }

  override def supportedCustomMetrics: Seq[StateStoreCustomMetric] = {
    metricStateOnCurrentVersionSizeBytes :: metricLoadedMapCacheHit :: metricLoadedMapCacheMiss ::
      Nil
  }

  override def toString(): String = {
    s"HDFSStateStoreProvider[" +
      s"id = (op=${stateStoreId.operatorId},part=${stateStoreId.partitionId}),dir = $baseDir]"
  }

  /* Internal fields and methods */

  @volatile private var stateStoreId_ : StateStoreId = _
  @volatile private var keySchema: StructType = _
  @volatile private var valueSchema: StructType = _
  @volatile private var storeConf: StateStoreConf = _
  @volatile private var hadoopConf: Configuration = _
  @volatile private var numberOfVersionsToRetainInMemory: Int = _
  @volatile private var numColsPrefixKey: Int = 0

  // TODO: The validation should be moved to a higher level so that it works for all state store
  // implementations
  @volatile private var isValidated = false

  private lazy val loadedMaps = new util.TreeMap[Long, HDFSBackedStateStoreMap](
    Ordering[Long].reverse)
  private lazy val baseDir = stateStoreId.storeCheckpointLocation()
  private lazy val fm = CheckpointFileManager.create(baseDir, hadoopConf)
  private lazy val sparkConf = Option(SparkEnv.get).map(_.conf).getOrElse(new SparkConf)

  private val loadedMapCacheHitCount: LongAdder = new LongAdder
  private val loadedMapCacheMissCount: LongAdder = new LongAdder

  private lazy val metricStateOnCurrentVersionSizeBytes: StateStoreCustomSizeMetric =
    StateStoreCustomSizeMetric("stateOnCurrentVersionSizeBytes",
      "estimated size of state only on current version")

  private lazy val metricLoadedMapCacheHit: StateStoreCustomMetric =
    StateStoreCustomSumMetric("loadedMapCacheHitCount",
      "count of cache hit on states cache in provider")

  private lazy val metricLoadedMapCacheMiss: StateStoreCustomMetric =
    StateStoreCustomSumMetric("loadedMapCacheMissCount",
      "count of cache miss on states cache in provider")

  private case class StoreFile(version: Long, path: Path, isSnapshot: Boolean)

  private def commitUpdates(
      newVersion: Long,
      map: HDFSBackedStateStoreMap,
      output: DataOutputStream): Unit = {
    synchronized {
      finalizeDeltaFile(output)
      putStateIntoStateCacheMap(newVersion, map)
    }
  }

  /**
   * Get iterator of all the data of the latest version of the store.
   * Note that this will look up the files to determined the latest known version.
   */
  private[state] def latestIterator(): Iterator[UnsafeRowPair] = synchronized {
    val versionsInFiles = fetchFiles().map(_.version).toSet
    val versionsLoaded = loadedMaps.keySet.asScala
    val allKnownVersions = versionsInFiles ++ versionsLoaded
    if (allKnownVersions.nonEmpty) {
      loadMap(allKnownVersions.max).iterator()
    } else Iterator.empty
  }

  /** This method is intended to be only used for unit test(s). DO NOT TOUCH ELEMENTS IN MAP! */
  private[state] def getLoadedMaps(): util.SortedMap[Long, HDFSBackedStateStoreMap] = synchronized {
    // shallow copy as a minimal guard
    loadedMaps.clone().asInstanceOf[util.SortedMap[Long, HDFSBackedStateStoreMap]]
  }

  private def putStateIntoStateCacheMap(
      newVersion: Long,
      map: HDFSBackedStateStoreMap): Unit = synchronized {
    if (numberOfVersionsToRetainInMemory <= 0) {
      if (loadedMaps.size() > 0) loadedMaps.clear()
      return
    }

    while (loadedMaps.size() > numberOfVersionsToRetainInMemory) {
      loadedMaps.remove(loadedMaps.lastKey())
    }

    val size = loadedMaps.size()
    if (size == numberOfVersionsToRetainInMemory) {
      val versionIdForLastKey = loadedMaps.lastKey()
      if (versionIdForLastKey > newVersion) {
        // this is the only case which we can avoid putting, because new version will be placed to
        // the last key and it should be evicted right away
        return
      } else if (versionIdForLastKey < newVersion) {
        // this case needs removal of the last key before putting new one
        loadedMaps.remove(versionIdForLastKey)
      }
    }

    loadedMaps.put(newVersion, map)
  }

  /** Load the required version of the map data from the backing files */
  private def loadMap(version: Long): HDFSBackedStateStoreMap = {

    // Shortcut if the map for this version is already there to avoid a redundant put.
    val loadedCurrentVersionMap = synchronized { Option(loadedMaps.get(version)) }
    if (loadedCurrentVersionMap.isDefined) {
      loadedMapCacheHitCount.increment()
      return loadedCurrentVersionMap.get
    }

    logWarning(s"The state for version $version doesn't exist in loadedMaps. " +
      "Reading snapshot file and delta files if needed..." +
      "Note that this is normal for the first batch of starting query.")

    loadedMapCacheMissCount.increment()

    val (result, elapsedMs) = Utils.timeTakenMs {
      val snapshotCurrentVersionMap = readSnapshotFile(version)
      if (snapshotCurrentVersionMap.isDefined) {
        synchronized { putStateIntoStateCacheMap(version, snapshotCurrentVersionMap.get) }
        return snapshotCurrentVersionMap.get
      }

      // Find the most recent map before this version that we can.
      // [SPARK-22305] This must be done iteratively to avoid stack overflow.
      var lastAvailableVersion = version
      var lastAvailableMap: Option[HDFSBackedStateStoreMap] = None
      while (lastAvailableMap.isEmpty) {
        lastAvailableVersion -= 1

        if (lastAvailableVersion <= 0) {
          // Use an empty map for versions 0 or less.
          lastAvailableMap = Some(HDFSBackedStateStoreMap.create(keySchema, numColsPrefixKey))
        } else {
          lastAvailableMap =
            synchronized { Option(loadedMaps.get(lastAvailableVersion)) }
              .orElse(readSnapshotFile(lastAvailableVersion))
        }
      }

      // Load all the deltas from the version after the last available one up to the target version.
      // The last available version is the one with a full snapshot, so it doesn't need deltas.
      val resultMap = HDFSBackedStateStoreMap.create(keySchema, numColsPrefixKey)
      resultMap.putAll(lastAvailableMap.get)
      for (deltaVersion <- lastAvailableVersion + 1 to version) {
        updateFromDeltaFile(deltaVersion, resultMap)
      }

      synchronized { putStateIntoStateCacheMap(version, resultMap) }
      resultMap
    }

    logDebug(s"Loading state for $version takes $elapsedMs ms.")

    result
  }

  private def writeUpdateToDeltaFile(
      output: DataOutputStream,
      key: UnsafeRow,
      value: UnsafeRow): Unit = {
    val keyBytes = key.getBytes()
    val valueBytes = value.getBytes()
    output.writeInt(keyBytes.size)
    output.write(keyBytes)
    output.writeInt(valueBytes.size)
    output.write(valueBytes)
  }

  private def writeRemoveToDeltaFile(output: DataOutputStream, key: UnsafeRow): Unit = {
    val keyBytes = key.getBytes()
    output.writeInt(keyBytes.size)
    output.write(keyBytes)
    output.writeInt(-1)
  }

  private def finalizeDeltaFile(output: DataOutputStream): Unit = {
    output.writeInt(-1)  // Write this magic number to signify end of file
    output.close()
  }

  private def updateFromDeltaFile(version: Long, map: HDFSBackedStateStoreMap): Unit = {
    val fileToRead = deltaFile(version)
    var input: DataInputStream = null
    val sourceStream = try {
      fm.open(fileToRead)
    } catch {
      case f: FileNotFoundException =>
        throw new IllegalStateException(
          s"Error reading delta file $fileToRead of $this: $fileToRead does not exist", f)
    }
    try {
      input = decompressStream(sourceStream)
      var eof = false

      while(!eof) {
        val keySize = input.readInt()
        if (keySize == -1) {
          eof = true
        } else if (keySize < 0) {
          throw QueryExecutionErrors.failedToReadDeltaFileError(fileToRead, toString(), keySize)
        } else {
          val keyRowBuffer = new Array[Byte](keySize)
          ByteStreams.readFully(input, keyRowBuffer, 0, keySize)

          val keyRow = new UnsafeRow(keySchema.fields.length)
          keyRow.pointTo(keyRowBuffer, keySize)

          val valueSize = input.readInt()
          if (valueSize < 0) {
            map.remove(keyRow)
          } else {
            val valueRowBuffer = new Array[Byte](valueSize)
            ByteStreams.readFully(input, valueRowBuffer, 0, valueSize)
            val valueRow = new UnsafeRow(valueSchema.fields.length)
            // If valueSize in existing file is not multiple of 8, floor it to multiple of 8.
            // This is a workaround for the following:
            // Prior to Spark 2.3 mistakenly append 4 bytes to the value row in
            // `RowBasedKeyValueBatch`, which gets persisted into the checkpoint data
            valueRow.pointTo(valueRowBuffer, (valueSize / 8) * 8)
            if (!isValidated) {
              StateStoreProvider.validateStateRowFormat(
                keyRow, keySchema, valueRow, valueSchema, storeConf)
              isValidated = true
            }
            map.put(keyRow, valueRow)
          }
        }
      }
    } finally {
      if (input != null) input.close()
    }
    logInfo(s"Read delta file for version $version of $this from $fileToRead")
  }

  private def writeSnapshotFile(version: Long, map: HDFSBackedStateStoreMap): Unit = {
    val targetFile = snapshotFile(version)
    var rawOutput: CancellableFSDataOutputStream = null
    var output: DataOutputStream = null
    try {
      rawOutput = fm.createAtomic(targetFile, overwriteIfPossible = true)
      output = compressStream(rawOutput)
      val iter = map.iterator()
      while(iter.hasNext) {
        val entry = iter.next()
        val keyBytes = entry.key.getBytes()
        val valueBytes = entry.value.getBytes()
        output.writeInt(keyBytes.size)
        output.write(keyBytes)
        output.writeInt(valueBytes.size)
        output.write(valueBytes)
      }
      output.writeInt(-1)
      output.close()
    } catch {
      case e: Throwable =>
        cancelDeltaFile(compressedStream = output, rawStream = rawOutput)
        throw e
    }
    logInfo(s"Written snapshot file for version $version of $this at $targetFile")
  }

  /**
   * Try to cancel the underlying stream and safely close the compressed stream.
   *
   * @param compressedStream the compressed stream.
   * @param rawStream the underlying stream which needs to be cancelled.
   */
  private def cancelDeltaFile(
      compressedStream: DataOutputStream,
      rawStream: CancellableFSDataOutputStream): Unit = {
    try {
      if (rawStream != null) rawStream.cancel()
      IOUtils.closeQuietly(compressedStream)
    } catch {
      case e: FSError if e.getCause.isInstanceOf[IOException] =>
        // Closing the compressedStream causes the stream to write/flush flush data into the
        // rawStream. Since the rawStream is already closed, there may be errors.
        // Usually its an IOException. However, Hadoop's RawLocalFileSystem wraps
        // IOException into FSError.
    }
  }

  private def readSnapshotFile(version: Long): Option[HDFSBackedStateStoreMap] = {
    val fileToRead = snapshotFile(version)
    val map = HDFSBackedStateStoreMap.create(keySchema, numColsPrefixKey)
    var input: DataInputStream = null

    try {
      input = decompressStream(fm.open(fileToRead))
      var eof = false

      while (!eof) {
        val keySize = input.readInt()
        if (keySize == -1) {
          eof = true
        } else if (keySize < 0) {
          throw QueryExecutionErrors.failedToReadSnapshotFileError(
            fileToRead, toString(), s"key size cannot be $keySize")
        } else {
          val keyRowBuffer = new Array[Byte](keySize)
          ByteStreams.readFully(input, keyRowBuffer, 0, keySize)

          val keyRow = new UnsafeRow(keySchema.fields.length)
          keyRow.pointTo(keyRowBuffer, keySize)

          val valueSize = input.readInt()
          if (valueSize < 0) {
            throw QueryExecutionErrors.failedToReadSnapshotFileError(
              fileToRead, toString(), s"value size cannot be $valueSize")
          } else {
            val valueRowBuffer = new Array[Byte](valueSize)
            ByteStreams.readFully(input, valueRowBuffer, 0, valueSize)
            val valueRow = new UnsafeRow(valueSchema.fields.length)
            // If valueSize in existing file is not multiple of 8, floor it to multiple of 8.
            // This is a workaround for the following:
            // Prior to Spark 2.3 mistakenly append 4 bytes to the value row in
            // `RowBasedKeyValueBatch`, which gets persisted into the checkpoint data
            valueRow.pointTo(valueRowBuffer, (valueSize / 8) * 8)
            if (!isValidated) {
              StateStoreProvider.validateStateRowFormat(
                keyRow, keySchema, valueRow, valueSchema, storeConf)
              isValidated = true
            }
            map.put(keyRow, valueRow)
          }
        }
      }
      logInfo(s"Read snapshot file for version $version of $this from $fileToRead")
      Some(map)
    } catch {
      case _: FileNotFoundException =>
        None
    } finally {
      if (input != null) input.close()
    }
  }

  /** Perform a snapshot of the store to allow delta files to be consolidated */
  private def doSnapshot(): Unit = {
    try {
      val (files, e1) = Utils.timeTakenMs(fetchFiles())
      logDebug(s"fetchFiles() took $e1 ms.")

      if (files.nonEmpty) {
        val lastVersion = files.last.version
        val deltaFilesForLastVersion =
          filesForVersion(files, lastVersion).filter(_.isSnapshot == false)
        synchronized { Option(loadedMaps.get(lastVersion)) } match {
          case Some(map) =>
            if (deltaFilesForLastVersion.size > storeConf.minDeltasForSnapshot) {
              val (_, e2) = Utils.timeTakenMs(writeSnapshotFile(lastVersion, map))
              logDebug(s"writeSnapshotFile() took $e2 ms.")
            }
          case None =>
            // The last map is not loaded, probably some other instance is in charge
        }
      }
    } catch {
      case NonFatal(e) =>
        logWarning(s"Error doing snapshots for $this", e)
    }
  }

  /**
   * Clean up old snapshots and delta files that are not needed any more. It ensures that last
   * few versions of the store can be recovered from the files, so re-executed RDD operations
   * can re-apply updates on the past versions of the store.
   */
  private[state] def cleanup(): Unit = {
    try {
      val (files, e1) = Utils.timeTakenMs(fetchFiles())
      logDebug(s"fetchFiles() took $e1 ms.")

      if (files.nonEmpty) {
        val earliestVersionToRetain = files.last.version - storeConf.minVersionsToRetain
        if (earliestVersionToRetain > 0) {
          val earliestFileToRetain = filesForVersion(files, earliestVersionToRetain).head
          val filesToDelete = files.filter(_.version < earliestFileToRetain.version)
          val (_, e2) = Utils.timeTakenMs {
            filesToDelete.foreach { f =>
              fm.delete(f.path)
            }
          }
          logDebug(s"deleting files took $e2 ms.")
          logInfo(s"Deleted files older than ${earliestFileToRetain.version} for $this: " +
            filesToDelete.mkString(", "))
        }
      }
    } catch {
      case NonFatal(e) =>
        logWarning(s"Error cleaning up files for $this", e)
    }
  }

  /** Files needed to recover the given version of the store */
  private def filesForVersion(allFiles: Seq[StoreFile], version: Long): Seq[StoreFile] = {
    require(version >= 0)
    require(allFiles.exists(_.version == version))

    val latestSnapshotFileBeforeVersion = allFiles
      .filter(_.isSnapshot)
      .takeWhile(_.version <= version)
      .lastOption
    val deltaBatchFiles = latestSnapshotFileBeforeVersion match {
      case Some(snapshotFile) =>

        val deltaFiles = allFiles.filter { file =>
          file.version > snapshotFile.version && file.version <= version
        }.toList
        verify(
          deltaFiles.size == version - snapshotFile.version,
          s"Unexpected list of delta files for version $version for $this: $deltaFiles"
        )
        deltaFiles

      case None =>
        allFiles.takeWhile(_.version <= version)
    }
    latestSnapshotFileBeforeVersion.toSeq ++ deltaBatchFiles
  }

  /** Fetch all the files that back the store */
  private def fetchFiles(): Seq[StoreFile] = {
    val files: Seq[FileStatus] = try {
      fm.list(baseDir)
    } catch {
      case _: java.io.FileNotFoundException =>
        Seq.empty
    }
    val versionToFiles = new mutable.HashMap[Long, StoreFile]
    files.foreach { status =>
      val path = status.getPath
      val nameParts = path.getName.split("\\.")
      if (nameParts.size == 2) {
        val version = nameParts(0).toLong
        nameParts(1).toLowerCase(Locale.ROOT) match {
          case "delta" =>
            // ignore the file otherwise, snapshot file already exists for that batch id
            if (!versionToFiles.contains(version)) {
              versionToFiles.put(version, StoreFile(version, path, isSnapshot = false))
            }
          case "snapshot" =>
            versionToFiles.put(version, StoreFile(version, path, isSnapshot = true))
          case _ =>
            logWarning(s"Could not identify file $path for $this")
        }
      }
    }
    val storeFiles = versionToFiles.values.toSeq.sortBy(_.version)
    logDebug(s"Current set of files for $this: ${storeFiles.mkString(", ")}")
    storeFiles
  }

  private def compressStream(outputStream: DataOutputStream): DataOutputStream = {
    val compressed = CompressionCodec.createCodec(sparkConf, storeConf.compressionCodec)
      .compressedOutputStream(outputStream)
    new DataOutputStream(compressed)
  }

  private def decompressStream(inputStream: DataInputStream): DataInputStream = {
    val compressed = CompressionCodec.createCodec(sparkConf, storeConf.compressionCodec)
      .compressedInputStream(inputStream)
    new DataInputStream(compressed)
  }

  private def deltaFile(version: Long): Path = {
    new Path(baseDir, s"$version.delta")
  }

  private def snapshotFile(version: Long): Path = {
    new Path(baseDir, s"$version.snapshot")
  }

  private def verify(condition: => Boolean, msg: String): Unit = {
    if (!condition) {
      throw new IllegalStateException(msg)
    }
  }
}


text:
Analytics Zoo Orca BigDL Estimator provides a set APIs for running BigDL model on Spark in a distributed fashion.

---

### Orca BigDL Estimator

Orca BigDL Estimator is an estimator to do BigDL training/evaluation/prediction on Spark in a distributed fashion.

It can support various data types, like XShards, Spark DataFrame, etc.

It supports BigDL backend in the unified APIs.

### Create Estimator from BigDL Model

You can create Orca BigDL Estimator with BigDL model.

```
from zoo.orca.learn.bigdl import Estimator
Estimator.from_bigdl(*, model, loss=None, optimizer=None, feature_preprocessing=None,
                   label_preprocessing=None, model_dir=None)
```
* `model`: BigDL Model to be trained.
* `optimizer`: BigDL optimizer.
* `loss`: BigDL criterion.
* `feature_preprocessing`: Used if the input data in fit function is Spark DataFrame.The param converts the data in feature column to a Tensor or to a Sample directly. It expects a List of Int as the size of the converted Tensor, or a Preprocessing[F, Tensor[T]]
    If a List of Int is set as feature_preprocessing, it can only handle the case that feature column contains the following data types: Float, Double, Int, Array[Float], Array[Double], Array[Int] and MLlib Vector. The feature data are converted to Tensors with the specified sizes before sending to the model. Internally, a SeqToTensor is generated according to the size, and used as the feature_preprocessing.
    Alternatively, user can set feature_preprocessing as Preprocessing[F, Tensor[T]] that transforms the feature data to a Tensor[T]. Some pre-defined Preprocessing are provided in package zoo.feature. Multiple Preprocessing can be combined as a ChainedPreprocessing.
    The feature_preprocessing will also be copied to the generated NNModel and applied to feature column during transform.
* `label_preprocessing`: Similar to feature_preprocessing, but applies to Label data.
* `model_dir`: The path to save model. During the training, if checkpoint_trigger is defined and triggered, the model will be saved to model_dir.

### Train BigDL model with orca BigDL Estimator
After an Estimator is created, you can call estimator API to train BigDL model:
```
fit(self, data, epochs, batch_size=32, feature_cols="features", label_cols="label",
    caching_sample=True, validation_data=None, validation_trigger=None,
    validation_metrics=None, checkpoint_trigger=None)
```
* `data`: Training data. SparkXShard and Spark DataFrame are supported.
* `epochs`: (int) Number of epochs to train the model.
* `batch_size`: (int) Batch size used for training. Default: 32.
* `feature_cols`: (string or list of string) Feature column name(s) of data. Only used when data is a Spark DataFrame. Default: "features".
* `label_cols`: (string or list of string) Label column name(s) of data. Only used when data is a Spark DataFrame. Default: "label".
* `caching_sample`: (Boolean) Whether to cache the Samples after preprocessing. Default: True.
* `validation_data`: Validation data. SparkXShard and Spark DataFrame are supported. Default: None.
* `validation_trigger`: Orca Trigger to validate model.
* `validation_metrics`: Orca validation methods.
* `checkpoint_trigger`: Orca Trigger to set a checkpoint.

### Inference with orca BigDL Estimator
After training or loading trained model, you can call estimator API to inference:
```
predict(self, data, batch_size=4, feature_cols="features", sample_preprocessing=None)
```
* `data`: Inference data. SparkXShard and Spark DataFrame are supported.
* `batch_size`: (int) Batch size used for inference. Default: 4.
* `feature_cols`:  (string or list of string) Feature column name(s) of data. Only used when data is a Spark DataFrame.
* `sample_preprocessing`: Used if the input data in predict function is Spark DataFrame. The user defined sample_preprocessing will directly compose Sample according to user-specified Preprocessing.

### Evaluate model
After Training, you can call estimator API to evaluate BigDL model:
```
evaluate(self, data, batch_size=32, feature_cols=None, label_cols=None, validation_metrics=None)
```
* `data`: Validation data. SparkXShard and Spark DataFrame are supported.
* `batch_size`: Batch size used for evaluation. Only used when data is a SparkXShard.
* `feature_cols`: (string or list of string) Feature column name(s) of data. Only used when data is a Spark DataFrame. Default: None.
* `label_cols`: (string or list of string) Label column name(s) of data. Only used when data is a Spark DataFrame. Default: None.
* `validation_metrics`: Orca validation methods.

### Get model
You can get model using `get_model(self)`

### Save model
You can save model using `save(self, model_path)`
* `model_path`: (str) Path to model saved folder.

### Load model
You can load saved model using
```
load(self, checkpoint, optimizer=None, loss=None, feature_preprocessing=None,
             label_preprocessing=None, model_dir=None, is_checkpoint=False):
```
* `checkpoint`: (str) Path to target checkpoint file or saved model folder.
* `optimizer`: BigDL optimizer.
* `loss`: BigDL criterion.
* `feature_preprocessing`: Used if the input data in fit function is Spark DataFrame.The param converts the data in feature column to a Tensor or to a Sample directly. It expects a List of Int as the size of the converted Tensor, or a Preprocessing[F, Tensor[T]]
    If a List of Int is set as feature_preprocessing, it can only handle the case that feature column contains the following data types: Float, Double, Int, Array[Float], Array[Double], Array[Int] and MLlib Vector. The feature data are converted to Tensors with the specified sizes before sending to the model. Internally, a SeqToTensor is generated according to the size, and used as the feature_preprocessing.
    Alternatively, user can set feature_preprocessing as Preprocessing[F, Tensor[T]] that transforms the feature data to a Tensor[T]. Some pre-defined Preprocessing are provided in package zoo.feature. Multiple Preprocessing can be combined as a ChainedPreprocessing.
    The feature_preprocessing will also be copied to the generated NNModel and applied to feature column during transform.
* `label_preprocessing`: Similar to feature_preprocessing, but applies to Label data.
* `model_dir`: The path to save model. During the training, if checkpoint_trigger is defined and triggered, the model will be saved to model_dir.
* `is_checkpoint`: (boolean) Whether load BigDL saved model or checkpoint.

### Set TensorBoard & get Training and Validation Summary
During training and validation, Orca BigDL Estimator would save summary data to specified log_dir. This data can be visualized in TensorBoard, or you can use estimator's APIs to retrieve it. You can set the logdir and app name with such API:
```
set_tensorboard(log_dir, app_name)
```
* `log_dir`: The base directory path to store training and validation logs.
* `app_name`: The name of the application.

This method sets summary information during the training process for visualization purposes. Saved summary can be viewed via TensorBoard. In order to take effect, it needs to be called before fit.

Training summary will be saved to 'log_dir/app_name/train' and validation summary (if any) will be saved to 'log_dir/app_name/validation'.

You can get Training summary with `get_train_summary(self, tag=None)` and Validation summary with `get_validation_summary(self, tag=None)`.

### Clear gradient clipping
You can clear gradient clipping parameters using `clear_gradient_clipping(self)`. In this case, gradient clipping will not be applied.
**Note:** In order to take effect, it needs to be called before fit.

### Set constant gradient clipping
You can Set constant gradient clipping during the training process using `set_constant_gradient_clipping(self, min, max)`.
* `min`: The minimum value to clip by.
* `max`: The maximum value to clip by.
**Note:** In order to take effect, it needs to be called before fit.

### Set clip gradient to a maximum L2-Norm
You can set clip gradient to a maximum L2-Norm during the training process using `set_l2_norm_gradient_clipping(self, clip_norm)`.
* `clip_norm`: Gradient L2-Norm threshold.
**Note:** In order to take effect, it needs to be called before fit.


text:
<META http-equiv="content-type" content="text/html;charset=iso-8859-9"/><meta name="viewport" content="width=device-width, initial-scale=1"><title>ac</title> <link rel="stylesheet"  href="css/jquery.mobile-1.2.0.css" />

KAYYUM-Ý CÝHAN MUHAMMED SEYFULLAH

Geri

Evliyanýn meþhurlarýndan. Ýmam-ý Rabbani hazretlerinin torunlarýndandýr. Ýsmi Muhammed,
babasýnýnki Gulam Muhammed Ma'sum'dur. 1743 (H.1156) senesinde doðdu.

Babasý Gulam Muhammed Ma'sum-i Sani, Kayyum-i Cihan Muhammed'in doðmadan önce
gördüðü rüyayý
þöyle anlatmýþtýr: Muhammed Seyfullah doðmadan bir gün önce, rüyamda
Peygamber efendimizi görüp, ziyaret etmekle þereflenmiþtim. Resulullah efendimiz bana

müjde vererek þöyle buyurdu: "Yarýn senin bir oðlun doðacak, görülmemiþ bir evlad ve
Allahü tealanýn sevgili kullarýndan olacaktýr. Dedelerinin; ecdadýnýn nisbetine varis olacaktýr.
Alemi nurla dolduracak, onun geliþinin çok bereketli olacaðýný bil ve ismini benim ismimden
koy." Kutbu'l-aktab Gulam Muhammed Ma'sum hazretleri, bu rüyayý gördükten sonra
uyanýp, abdest aldý ve sabah namazýný kýlmak için dergahýna gitti. Namazdan sonra murakabe
halinde otururken, Kabe'yi gördü ve çok mübarek bir oðlunun dünyaya geldiði müjdesini
aldý. Müceddidiyye ve Ma'sumiyye bahçesinin bu yeni fidanýnýn kulaðýna ezan ve ikamet
okuduðu zaman, bu yeni doðmuþ bebek, yanýnda bulunanlarýn da hayretle görüp duyduklarý
bir þekilde tekbir getirdi. Ýsmi konurken tekbir getirdiðini iþitenler, babasýna; "Onun tekbir
getirdiðini iþittik." dediler. Babasý, ona hizmet etmelerini söyledikten sonra; "Bu çocuk
asrýnýn bir tanesi olacak, görülmemiþ nadir iþler yapacak, onun irþad nurlarý alemi dolduracak
ve insanlar bahtiyar olacak. Bizim hayatýmýz ona vesile olmak içindir. Ýþte þimdi bu nadir
evlad doðdu. Maksad hasýl oldu." demiþtir.

Yine onun için þöyle anlatmýþtýr: "Bu oðlum diðer çocuklar gibi aðlamazdý. Önce Allahü
tealanýn ismini zikreder, üç defa gayet açýk bir þekilde "Allah" derdi. Bundan sonra aðlardý.
Bu çocukda öyle haller görüyorum ki, onun sýrlarýndan olan bu hallerin çoðunu kimse
bilmiyor!" Babasý onun yetiþmesi için yüksek teveccühlerde bulundu. Onu Müceddidi
yolunun nurlarýna ve sýrlarýna kavuþturdu.

Kayyum-i Cihan Muhammed Seyfullah hazretleri, önce babasý Gulam Muhammed Ma'sum-i
Sani'nin teveccüh ve feyzleriyle yetiþti. Bunu kendisi þöyle anlatmýþtýr: "Hakikatý arayanlarýn
rehberi olan babam hayattayken, ben her ne kadar küçüksem de babamýn feyzlerine
kavuþtum. Çünkü, babamýn feyzleri ve bereketli teveccühleri, büyük-küçük, genç-ihtiyar,
diri-ölü herkese ulaþýyordu. Benimle daha baþka bir þefkat ve dikkat ile ilgilendi. Çok
teveccühte bulundu. Bana yüksek müjdeler ve iþaretler verdi. Hakkýmda müjdelediði ve iþaret
ettiði þeylerin hepsine kavuþtum. Babamýn benim hakkýmda verdiði müjdeleri iþiten ahbabý
ve seçilmiþ eshabý yýllar sonra zamaný geldikçe benim o nimetlere kavuþtuðumu görünce
þaþýrdýlar."

Kayyum-i Cihan Muhammed Seyfullah hazretleri, daha çocukluðunda þahid olduðu
hadiseleri þöyle anlatmýþtýr: Çocukluðumda acayib haller görür ve yaþardým. Büyüklerin
ruhlarýný ve cinlerin salihlerini görürdüm. Çocukluk günlerimde bir defasýnda akranlarýmdan
bir grup çocukla Lahor þehrine gitmiþtik. O günlerde çevreye gayr-i müslimler gelip
yerleþmiþlerdi. Ben bir muz aðacýna çýkýp, muzlarýn iyilerini topluyor ve çocuklara
atýyordum. Aðacýn altýndaki çocuklar gayr-i müslim atlýlarýn yanýmýza geldiðini görerek
kaçmýþlar, ben aðacýn üzerinden; "Attýðým muzlarý toplayýnýz." diye baðýrmýþtým. Bir de
baktým hepsi gitmiþ. Bu arada aðacýn altýnda baþka çocuklar gördüm. Aðýzlarýndan ateþ
çýkýyordu. Onlara size ne oluyor da aðzýnýzdan ateþ saçýlýyor dedim ve hayret ettim. Onlar;
"Biz cinlerin çocuklarýyýz. Buraya seni korumak için toplandýk. Öbür çocuklar süvarilerden
korkup kaçtýlar!" dediler.

Yine çocukluðumda bir gün ormana gitmiþtim. Yanýmda kimse yoktu. Ormanda dolaþýrken
bir arslanýn peþimden geldiðini gördüm. Nereye gitsem peþimden takib etti ve asla bana
saldýrmadý, bir zarar vermedi. Sonra ormandan çýkýp gidinceye kadar benden ayrýlmadý.
Ormandan ayrýldým, arslan da peþimden ayrýldý. Onun, bana bekçilik yaptýðýný ve beni
koruduðunu anladým.

Yine birgün ata binmiþ gidiyordum. Hizmetçiler de peþimden geliyordu. Yolda at aniden
durdu ve yürümedi. Sebebini araþtýrýp çevreye baktým ve bir arslanýn üzerime saldýrmaya
hazýrlandýðýný gördüm. Tam karþýma çýkýnca dikkatlice baktým. Ben böyle bakar bakmaz
arslan geri dönüp sür'atle kaçtý. Allahü tealanýn beni koruduðunu, hýfz-ý ilahinin benimle
olduðunu anladým."

Kayyum-i Cihan Muhammed Seyfullah'ýn çocukluðunda, buna benzer pekçok hadiseler

geçmiþtir. Dört yaþýndayken babasý vefat etmiþtir. Babasýndan çok feyz alýp, onun vefatýndan
sonra da kardeþi Þah Gulam Muhammed'den feyz alarak yetiþmiþ, böylece büluð çaðýna
ulaþmýþtýr. Çocukluk halini anlatýrken: "Çocukluðumda herhangi bir hastalýða tutulsam,
hastalýðýn tedavisi ve ilacýný, babamýn feyzi ve bereketi bilirdim ve böylece tedavi olurdum."
buyurmuþtur. Babasýnýn vefatýndan sonra, yýllarca onun ayrýlýk ateþiyle yandý. Sonra aðabeyi
Þah Gulam Muhammed onu tasavvufta yetiþtirip, kemal derecelere ulaþtýrdý ve icazet verdi.

Kayyum-i Cihan Muhammed Seyfullah hazretleri, tasavvufta yetiþip icazet aldýktan sonra,
insanlara doðru yolu anlatmaya baþladý. Kýrk yaþýna yaklaþtýðý sýrada Türkistan'a gitti. Bu
sýrada tasavvuf hallerine gark olmuþ, yüksek derecelerde, kendinden geçmiþ bir haldeydi.
Kabil'e vardýklarýnda halleri pek yüksek derecelere ulaþmýþtý. Orada insanlara ilim yaymakla
meþgul oldu. O bölgede hizmetleri çok tesirli olup, insanlar sohbetine koþtular ve feyzleri ile
saadete kavuþtular. Hatta meþhur kumandanlar onu ziyaret etmek, sohbetinde bulunmak
arzusuyla bulunduðu yere gittiler.

Kayyum-i Cihan MuhammedSeyfullah hazretleri, her iþinde olduðu gibi nafile namazlarý
kýlma hususunda da sünnete uyardý. Namaz kýlarken namazýn tadil-i erkanýna, edeblerine
riayet eder, hudu', huþu' ve tumaninet içinde olurdu. Kýyamda ve secdede uzun müddet
dururdu. Kendinden geçmiþ, kalbi Allahü tealaya yönelmiþ, dünya düþüncelerinden tamamen
kurtulmuþ bir halde namaz kýlardý. Her hafta, peþi peþine olmak üzere; pazartesi, salý,
çarþamba ve perþembe günleri bazan da hafta boyunca oruç tutardý. Giyinme hususunda da
sünnete uyardý. Asla bid'at iþlemezdi. Hiçbir bid'ati beðenmez ve kabul etmezdi. Bid'at
sahiplerinden uzak durur, onlarý meclisine kabul etmezdi. Dünyaya düþkün olanlarý huzuruna
kabul etmez, zenginlerle görüþmezdi.

Kayyum-i Cihan hazretleri, bir gün yeðeninin evine gitmiþti. O sýrada, kapýda bir Hintli
duruyordu. O gelince kenara çekilip yol verdi. Bu sýrada Kayyum-ý Cihan hazretleri ona bir
nazar edip içeri girdi. Ýçeri girince; kapýdaki Hintliyi sordu. Bir ihtiyaç için geldiðini
söylediler. Bunun üzerine; "Yakýnda sohbetimize gelir." buyurdu. Bunu iþitenlerden biri;
"Hindunun müslüman olacaðýna dair müjde verildi." dedi. Bir müddet sonra o Hindu bir arsa
yüzünden biriyle hasým oldu. Meselenin halli için Ýslam kadýsýna baþ vurdu. Kadý huzurunda
konuþurken, sözlerinden müslüman olduðu anlaþýlýyordu. Daha sonra müslüman olduðunu
açýkladý ve Kayyum-ý Cihan hazretlerinin dergahýna gidip talebelerinden oldu.

Umdet-ül-Makamat kitabýnýn müellifi Hace Muhammed Fadl þöyle anlatmýþtýr: Evimizin
önünde bir söðüt aðacý, vardý. Bir ara bu aðaç kurudu. Bir gün Kayyum-i Cihan hazretleri
abdest alýrken, abdest aldýðý suyu bir kaba koydum. Niçin aldýðýmý sorunca; "Kurumuþ bir
söðüt aðacýmýz var. Onun dibine dökeceðim." dedim. "Eðer bu niyetle dökeceksen o aðaç
yeþerir." buyurdu. Götürüp söðüt aðacýnýn dibine döktüm. Onun bereketiyle yeþermesini arzu
ediyordum. Nihayet aðaç yeþerip taze dallar verdi. Kayyum-i Cihan hazretleri bir gün bu
söðüt aðacýna bakarak; "Bu aðaç biz hayatta olduðumuz müddetçe bazan kurur, bazan
yeþerir. Fakat biz vefat ettikten sonra tamamen kurur!" buyurdu. Gerçekten de buyurduðu
gibi oldu.

Kayyum-i Cihan hazretleri bir gün abdest aldý, hizmetçiye dönerek; Bu abdest suyunu sakla
ona ihtiyaç olacak." buyurdu. Hizmetçi; "Ne için lazým olacak?" deyince; Bir yýlanýn
zehirinin tesirini defetmek için!" buyurdu. SonraCuma namazýna gitti. Namazdan dönünce;
huzuruna zayýf ve bitkin bir kimse getirdiler ve; "Bunu yýlan soktu ölmek üzeredir. Bir de
küçük çocuðu var, babasýz, yetim kalacak! Bir duada ve yardýmda bulunun!" dediler.
Kayyum-i Cihan hazretleri Cuma namazýna gitmeden önce abdest alýrken hizmetçiye bir kaba
koymasýný söylediði abdest suyunu istedi. "Onu bunun için ayýrmýþtýk." buyurdu.Sonra
yýlanýn soktuðu kimseye verip, bir kýsmýný yýlanýn soktuðu yere sürmesini söyledi.
Buyurduðu gibi yapýnca hemen iyileþti.

Ömrünün son günlerinde, Peygamber efendimizin kabr-i þerifini ziyaret arzusuyla yandý.

Gençliðinde hacca gidip, ziyaret etmiþti. Hayatýnýn son zamanlarýnda Kabil'e gitti.Son
günlerinde ibadet ve taatlarýný pek ziyade arttýrmýþtý. Bu halini görüp ibadetleri çok
ziyadeleþtirdiniz denilince, tebessüm ederek; "Artýk ömür sona erdi, elden ne gelirse yapmak
lazým." buyurdu. Son günlerinde sohbetleri kalabalýktan taþmaya baþladý. Pekçok kimse onun
sohbetini bulunmaz bir ganimet bilerek feyzlerine kavuþuyordu. Vefatýndan önce sýtma
hastalýðýna tutulup, hastalýðý yedi gün þiddetle devam etti. Hastalýðý sýrasýnda alnýndan
terlerin aktýðý görülerek, þifaya kavuþacaðýnýza alamet denilince, tebessüm edip; "Ümid
ediyorum ki, ahiret þifasý hasýl olacak!" buyurdu. Son nefesleri verdiði sýralarda baþýnda
bulunanlar yüzündeki nurun git gide fazlalaþtýðýný görerek hayret ettiler. Bu haldeyken
baþýnda Yasin-i þerif suresinin okunmasýný emretti ve okundu. Kendisi de hep zikir ile
meþgul oluyordu. Bu haldeyken tebessüm ederek 1797 (H.1212) senesinde vefat
etti.Cenazesi yýkanýrken yanýnda bulunan Hacý Muhammed Fadl þöyle anlatmýþtýr: "Cenazesi
yýkanýrken ben ayak ucunda duruyordum. Yýkama iþi bitince yüzüyle bana karþý bir iþaret
yaparak ayaklarýný gösteriyordu. Ayak ucuna gidip baktýðýmda az bir yerin kuru kaldýðýný
gördük ve yýkadýk. Cenazesinde büyük bir cemaat toplanmýþtý.Kabre koymak üzere kabrine
inmiþtim. Mübarek yüzünü birazcýk kýbleye karþý çeviriyordum. Kendisi gayet yavaþ bir
hareketle yüzünü kýbleye çevirdi. Yüzünü açtým, yüzünde görülmedik bir nur parlýyordu ve
gittikçe artýyordu ki akýl onu anlayamaz."

Kabri, Ruþeni Hindi Mescidi yanýndadýr. Vefat etmeden önce bir gün bir yere oturup sohbet
etmiþti. Orada gayet güzel ve yeþil bir aðaç vardý. Aðaç altýnda oturulduðu sýrada; "Burasý
kamil ve mükemmil bir zatýn mekaný olacak." buyurmuþtu. Vefat edince kendisi iþaret ettiði o
yere defnedildi.

Kayyum-i Cihan Muhammed Seyfeddin hazretlerinin eserleri: 1) Adab-ül-Ýrþad: Bu eserine
Ma'den-ül-Esrar adý da verilmiþtir. 2) Mahzen-ül-Envar-i Ahmedi fi
Keþf-il-Esrar-il-Müceddidi, 3) Çihar Cuy (Dört Nehir).

GERÇEK TABÝB

Bir defasýnda Timur Þah hastalandý ve bir türlü çare bulamadý. Nihayet Kayyum-i Cihan hazretlerine
haber göndererek; "Tabibler tedavisinden aciz kaldýlar, bir çare bulamadýlar! Þimdi sizin himmetinizi
beklemektedir." dediler.

Bu haber üzerine Kayyum-i Cihan hazretleri hemen; "Esferze bitkisinden bir miktar ilaç yapýp üç gün
yutsun, inþaallahü teala þifa bulacak." buyurdu. Bu durumu tabibler duyunca hepsi birden ittifak
halinde; "Onun hastalýðý soðuk sebebiyle olmuþtur. Eðer "Esferze" bitkisinden yapýlan ilaç verilirse
ölümüne sebeb olur. Fakat madem ki, Kayyum-i Cihan böyle buyurmuþ tecrübe edilsin." dediler.
Bunun üzerine Timur Þah'ýn tabibi olan en meþhur tabib; "Tecrübe gerekmez. Bu, canla oynamaktýr."
dedi. Tabiblerin endiþesi ve bu hususdaki dedikodularý Kayyum-i Cihan hazretlerine bildirilince, bir
miktar "Esferze" bitkisi istedi. Kendi eliyle ilaç hazýrlayýp, Timur Þaha gönderdi ve; "Bu bitkide üç
çeþit özellik vardýr. Hiç endiþelenmesin. Üç gün bu ilacý
kullansýn, sýhhate kavuþacak." buyurdu.
Timur Þah, Kayyum-i Cihan hazretlerinin tavsiye ettiði ve eliyle yaparak gönderdiði ilacý üç gün
kullandý. Sonunda sýhhate kavuþtu. Timur Þah ve ilacýn kullanýlmasýna mani olmak isteyen tabibler
hayret ettiler. Kayyum-i Cihan hazretlerinin ilim ve marifet sahibi bir veli olduðunu anlayýp sohbetine
katýldýlar.

1) Umdet-ül-Makamat; s.445

2) Ýslam Alimleri Ansiklopedisi; c.18, s.62


text:
Passport-Shibboleth
====================

This is a [Shibboleth](http://shibboleth.net) authentication provider for [Passport](http://passportjs.org/)

The code was originally based on @drstearns [implementation](https://github.com/drstearns/passport-uwshib)

## Installation

```
npm install passport-shibboleth --save
```

## Usage

### Configure strategy

```
passport.use(new ShibbolethStrategy({

  },
  function(profile, done) {
    findByEmail(profile.email, function(err, user) {
      if (err) {
        return done(err);
      }
      return done(null, user);
    });
  });
);
```

### Parameters

* `entryPoint` - Full SAML2 redirect url (Eg: https://test-idp.ukfederation.org.uk/idp/profile/SAML2/Redirect/SSO)
* `authUrl` - If entryPoint is not known, use this parameter (Eg: https://test-idp.ukfederation.org.uk/idp/shibboleth - Note. This is also called as entityId)
* `publicCert or publicCertFile` - Public certificate string in pem format (Eg: MIIFUjCC ...)
* `entityId` - This is you the service provider (SP). Can be domain name.
* `callbackUrl` - Url to calback after authentication. If not specified /saml/callback will be appended to the domain and used
* `privateKey or privateKeyFile` - Private key in pem format

### Using attributes - Privacy note

NOTE: We are not legal experts and this is not a legal advice.

Data protection law and the institutional obligation to preserve user privacy both require that information idenfifying individuals only be exchanged when strictly necessary. This means that if you are storing attributes other than eduPersonScopedAffiliation and eduPersonTargetedID (Eg: eduPersonPrincipalName), you should have a privacy policy in place.