Клуб всё?

В конце октября может прекратить существование клуб «Виндсерф» имени Владимира Маслаченко, в свое время ставший первым в стране и работающий уже 27 лет. Дескать, не вписывается спортивный объект в новую концепцию благоустройства столичного озера.
http://www.mk.ru/moscow/2015/08/24/pervaya-v-strane-shkola-vindserfinga-pod-ugrozoy-zakrytiya.html

https://youtu.be/hlz2Y8ZLG5Q

Большие люди решили -- облагородить Строгино и пойму. Исполнители нарисовали план. Строители построили.
А такие мелочи, как уникальная точка доступа к чудесам парусного спорта, клуб Виндсерф, никто и не заметил. У нас не принято проверять и согласовывать планы по месту. Зачем?

И ведь, ЧСХ, клуб занимает пространства всего ничего, пятачок 30х30 метров берега. А плотность людей там заметно выше, чем в любом другом месте поймы. Особенно, когда тепло и ветер дует. Но эти люди не интересуют биг боссов. Их интересует бюджет, план и лизнуть начальника.

Это ресторан можно построить почти в любом месте. А база парусного / водного спорта требует редкого сочетания параметров. А такая точка как клуб Виндсерф -- вообще уникальна. В черте города, на большом водоеме, недорого, в двухстах метрах от остановки трамвая.

Раньше можно было подъехать и на машине, что удобно чертовски, ибо снарягу и оборудование на трамвае возить и в руках таскать ну очень тяжело. Но в процессе облагораживания подъезд перекрыли, парковку заговняли торфом, не потрудившись даже пешеходных дорожек сделать до клуба. Ведь его в планах нет. А дальше и сам клуб сроют, за ними не заржавеет.
Ненавижу.

Дорогой Дмитрий Медведев, не вы ли предложили развивать парусный спорт? Ваши миньоны с вами не согласны.



original post http://vasnake.blogspot.com/2015/08/blog-post_31.html

Week 3

Курс Scalable Machine Learning.

В прошлый раз я обрисовал материал второй недели обучения. Пощупали Apache Spark руками.

Теперь посмотрим, что было на третьей неделе.

Линейная регрессия и как ее обсчитывать распределенно.

WEEK 3: Linear Regression and Distributed Machine Learning Principles

Topics:

• Linear regression formulation and closed-form solution,
• distributed machine learning principles (related to computation, storage, and communication),
• gradient descent,
• quadratic features,
• grid search.

Lab 3: Millionsong Regression Pipeline.

Develop an end-to-end linear regression pipeline to predict the release year of a song given a set of audio features.

You will implement a gradient descent solver for linear regression, use Spark's machine Learning library ( mllib) to train additional models, tune models via grid search, improve accuracy using quadratic features, and visualize various intermediate results to build intuition.

Почему линейная регрессия? Потому, что это просто, удобно и, обычно, дает хорошие результаты. Плюс, другие методы внутри очень похожи на математику линейной регрессии.

Задача линейной регрессии (supervised learning problem) заключается в отыскании такого вектора w, чтобы умножение этого вектора на вектор свойств (features) давало желаемый результат.

y = w0*x0 + w1*x1 + w2*x2 + w3*x3

То есть, надо решить задачу оптимизации, минимизировать потери

(y — yTrain)**2

Эту задачу можно решить аналитически, closed-form solution. Такое решение существует, если существует инверсная матрица

inv(Xtranspose * X)

w = inv(Xtranspose * X) * Xtranspose * y

На самом деле, это неинтересно, ибо такое решение имеет слишком много ограничений для применения в реальных БигДата условиях.

Нам нужен итеративный алгоритм, который можно порешать через вычисление кусочков. 

Но пока нам заливают про общие проблемы: overfitting, regularization, generalization and so on.

Чтобы избежать оверфиттинга, надо снизить сложность модели. Делается это через дополнительный коэффициент, переменную регуляризации.

Чем больше лямбда, тем ниже сложность модели и меньше риск оверфиттинга. Но и точность меньше.

Модель с регуляризацией называется Ridge Regression.

Чтобы подобрать подходящую лямбду, иными словами – найти оптимальный гиперпараметр модели, используется набор данных cross validation set. Как мы помним, начальный набор данных делится на три: training set, test set, cross validation set.

Меняя гиперпараметр, мы обучаем модель на training set и проверяем на cross validation set. Выбираем такой гиперпараметр, который на кросс-валидации дает наименьшую ошибку.

Метод поиска гиперпараметров называется grid search, потому как для двух гиперпараметров схема похожа на решетку.

И закончить описание регрессии можно способом оценки результата.

Оценка проводится через вычисление RMSE root-mean-square error.

Сумма квадратов разниц, поделенная на количество наблюдений. Да, проверка проводится на test set.

Дальше стало интереснее, пошла тема про особенности распределенного вычисления регресии.

Что нам стоит дом построить найти closed-form solution для линейной регрессии с наименьшей квадратичной ошибкой

w = inv(Xtranspose * X) * Xtranspose * y

Какие потребности во времени и хранилище?

Выходит так, что Xtranspose * X по времени занимает O(n * d**2)

а inv(matrix) уже O(d**3)

То есть, когда количество фичей (features) становится очень большим, считать closed-form solution уже совсем невыгодно.

In summary, we say that computing

the closed-form solution for linear regression

takes O of nd squared plus d cubed time.

С хранилищем не лучше.

Xtranspose * X и их инверсия занимают O(d**2) места,

тогда как X требует O(nd).

В общем, выходит так, что при большом количестве наблюдений n и малом количестве фич d, мы можем посчитать closed-form solution, распределив записи (n) по разным выч.узлам.

Фишка в том, что Xtranspose * X можно посчитать как сумму outer products между строками X. Поэтому строки можно раздать на разные ноды и там считать.

Но, в целом, это все имеет чисто академический интерес для нас. Ибо первое правило гласит: вычислительная и хранительная сложность должны быть линейны в терминах n, d.

Нам нужен другой алгоритм.

Кроме того, можно использовать такое свойство данных как разреженность (sparsity). Очень часто вектор содержит 99% нулей и только 1% значимых чисел.

Данные можно сжать и обрабатывать в сжатом виде.

Другой вариант, применить метод снижения размерности, типа PCA. И работать с сильно сокращенной версией данных.

Но всех спасти может только Gradient Descent. Алгоритм оптимизации (поиск минимума loss function) с линейной зависимостью от n,d.

Все, что требуется, это наличие одного минимума у оптимизируемой функции.

Thankfully for us, least squares regression, ridge regression,

and logistic regression, which we'll

be looking at later in this course,

all involve minimizing convex functions.

Направление движения к минимуму мы определяем по знаку производной;

Количество итераций алгоритма – гиперпараметр (можно подобрать через grid search);

Шаг движения alpha – тоже гиперпараметр.

На практике применяют постепенное уменьшение alpha

alpha[i] = alpha / (n * sqrt(i))

i – iteration number

n — training points (records)

This entire process can be concisely described

via the following Spark code snippet.

In this snippet, we perform a fixed number of iterations

as specified by numIters.

And on each iteration, we first compute

the iteration-specific step size and then compute the gradient

via a MapReduce operation.

Finally, we use the gradient descent update

rule to update w.

Now to conclude, gradient descent

has some nice favorable properties.

It is easily parallelized, and each iteration

is cheap enough to scale to the big N, big D setting.

Additionally, stochastic variance of gradient descent

can be used to further speed up the algorithm.

Но, что поделать, алгоритм медленно сходится и требует общения между нодами – передача параметров w на каждой итерации на каждую ноду.

В завершение описания Gradient Descent нам продемонстрировали важность таких вещей как локальность вычислений и минимизация сообщений.

Передавать данные дорого и чем дальше от процессора, тем дороже.

Mini-batch.

Поэтому мы держим RDD в памяти (используя директиву cache) и разрабатываем такие алгоритмы, чтобы распараллелить данные и вычисления, снизив до минимума количество коммуникаций между нодами.

Используем факт разреженности данных, разные оптимизации для снижения количества итераций.

Distributed iterative algorithms was both compute

and perform communication.

And in a bulk synchronous, in Bulk Synchronous Parallel

Systems or BSP systems like Apache Spark

we strictly alternate between computation and communication,

so that all worker nodes are synchronized.

Hence each iteration of an iterative algorithm

incurs some communication overhead.

In order to take advantage of parallel computation,

but reduce network cost, we'd like

to design algorithms that compute more and communicate

less.

В принципе, мы можем раздать по нодам данные, необходимые для полных вычислений и собрать обратно уже готовые результаты. Все итерации будут проведены над частями данных в рамках отдельных нод. Так мы максимально сократим коммуникации, но получим только приблизительные результаты, усредненные.

Компромиссный вариант, это mini-batch метод. На нодах вычисляются наборы итераций и так сокращается общее количество итераций с уменьшением коммуникций.

In parallel gradient descent, at each iteration

each worker simply computes its contribution to the gradient

and sends it back to the driver.

In contrast, at the beginning of each mini-batch iteration,

each worker starts with the same parameter vector,

but then performs a few steps of gradient descent locally.

After each of the k workers has performed its local updates,

these workers then send their updated parameter vectors

back to the driver and the driver

combines these different models, perhaps via averaging.

И, конечно, не надо забывать о network latency. Сообщения надо собирать в большие пакеты и отправлять весь пакет разом. Например, через раздачу на ноды сразу нескольких моделей для тренировки.

Про лабораторку расскжу в следующий раз.

original post http://vasnake.blogspot.com/2015/08/week-3.html

Лучше любой пропаганды

Красивые и громкие слова, это, конечно, заебись. Но реальную картину можно увидеть только опираясь на факты.

Например, на разные числа, характеризующие бюджет, на что он расходуется.

Вот, две говорящие головы, одна из них – Наталья Зубаревич. Она расскажет а вы послушайте, может поймете, что в Ресурсной Федерации происходит.

https://youtu.be/OoPcjpRtYE4

Некоторые предпочитают верить а не думать. Так многим проще, ответственности меньше, напрягаться не надо. Ну что-ж, свободные люди в свободной стране, имеют право. Только вот не надо забывать, что любые действия влекут за собой последствия. Даже если действие – отсутствие действия как такового.




original post http://vasnake.blogspot.com/2015/08/blog-post.html

Week 2, lab

Курс Scalable Machine Learning.

В прошлый раз я начал рассказывать о материале, который мы изучали на второй неделе – Introduction to Apache Spark. Но не закончил. Продолжаю.

Lab 2

Learning Apache Spark. ... you will learn about the Spark data model, transformations, and actions, and write a word counting program to count the words in all of Shakespeare's plays.

Как известно, лабораторки в этом курсе проводятся на личных виртмашинах, поднятых через Vagrant и VirtualBox.

vagrant up
http://localhost:8001/

Нотебуки там такие

https://github.com/spark-mooc/mooc-setup/blob/master/spark_tutorial_student.ipynb

https://github.com/spark-mooc/mooc-setup/blob/master/ML_lab2_word_count_student.ipynb

И понеслась.

Интересные топики в этой лабе:

– An introduction to using Apache Spark with the Python pySpark API running in the browser

– Using RDDs and chaining together transformations and actions

– Lambda functions

– Additional RDD actions

– Additional RDD transformations

– Caching RDDs and storage options

The following transformations will be covered:

• map(),
• mapPartitions(),
• mapPartitionsWithIndex(),
• filter(),
• flatMap(),
• reduceByKey(),
• groupByKey()

Трансформации, это преобразование одного RDD в другой, распределенно и лениво. На воркерах.

The following actions will be covered:

• first(),
• take(),
• takeSample(),
• takeOrdered(),
• collect(),
• count(),
• countByValue(),
• reduce(),
• top()

Действия (акции), это получение результата обсчета на драйвере.

Also covered:

cache(), unpersist(), id(), setName()

SparkContext – грубо говоря, коннект к кластеру, задающий параметры типа количества узлов и проч. В лабах этот контекст создается за кадром и доступен через переменную sc.

At a high level, every Spark application consists of a driver program that launches various parallel operations on executor Java Virtual Machines (JVMs) running either in a cluster or locally on the same machine.... When running locally, "PySparkShell" is the driver program. In all cases, this driver program contains the main loop for the program and creates distributed datasets on the cluster, then applies operations (transformations & actions) to those datasets

Driver programs access Spark through a SparkContext object, which represents a connection to a computing cluster. A Spark context object (sc) is the main entry point for Spark functionality

Управлятор Spark

You can view the details of your Spark application in the Spark web UI. When running locally you'll find it at localhost:4040. In the web UI, under the "Jobs" tab, you can see a list of jobs that have been scheduled or run.

Создадим RDD

data = xrange(1, 10001)
xrangeRDD = sc.parallelize(data, 8)
xrangeRDD.setName('My first RDD')

Проще пареной свеклы.

Теперь посмотрим на трансформации и действия

Теперь выполним трансформацию, создав новый RDD через уменьшение каждого элемента на 1.

def sub(value):
    return (value - 1)
subRDD = xrangeRDD.map(sub)

А теперь actions

print subRDD.collect()
print xrangeRDD.count()

Трансформация filter

def ten(value):
    if (value < 10):
        return True
    else:
        return False
filteredRDD = subRDD.filter(ten)
print filteredRDD.collect()

Теперь с лямбдами

lambdaRDD = subRDD.filter(lambda x: x < 10)
lambdaRDD.collect()
evenRDD = lambdaRDD.filter(lambda x: x % 2 == 0)
evenRDD.collect()

Другие действия.

Let's investigate the additional actions: first(), take(), top(), takeOrdered(), and reduce()

# Let's get the first element
print filteredRDD.first()
# The first 4
print filteredRDD.take(4)
# Note that it is ok to take more elements than the RDD has
print filteredRDD.take(12)
# Retrieve the three smallest elements
print filteredRDD.takeOrdered(3)
# Retrieve the five largest elements
print filteredRDD.top(5)
# Pass a lambda function to takeOrdered to reverse the order
filteredRDD.takeOrdered(4, lambda s: -s)
# Obtain Python's add function
from operator import add
# Efficiently sum the RDD using reduce
print filteredRDD.reduce(add)
# Sum using reduce with a lambda function
print filteredRDD.reduce(lambda a, b: a + b)
# Note that subtraction is not both associative and commutative
print filteredRDD.reduce(lambda a, b: a - b)
print filteredRDD.repartition(4).reduce(lambda a, b: a - b)
# While addition is
print filteredRDD.repartition(4).reduce(lambda a, b: a + b)

Here are two additional actions that are useful for retrieving information from an RDD: takeSample() andcountByValue()

# takeSample reusing elements
print filteredRDD.takeSample(withReplacement=True, num=6)
# takeSample without reuse
print filteredRDD.takeSample(withReplacement=False, num=6)
# Set seed for predictability
print filteredRDD.takeSample(withReplacement=False, num=6, seed=500)
# Try reruning this cell and the cell above -- the results from this cell will remain constant

# Create new base RDD to show countByValue
repetitiveRDD = sc.parallelize([1, 2, 3, 1, 2, 3, 1, 2, 1, 2, 3, 3, 3, 4, 5, 4, 6])
print repetitiveRDD.countByValue()

Трансформация flatMap нужна, если на входе RDD с элементами в виде коллекций, а нам надо плоский список.

The solution is to use a flatMap()transformation, flatMap() is similar to map(), except that with flatMap() each input item can be mapped to zero or more output elements.

# Let's create a new base RDD to work from
wordsList = ['cat', 'elephant', 'rat', 'rat', 'cat']
wordsRDD = sc.parallelize(wordsList, 4)

# Use map
singularAndPluralWordsRDDMap = wordsRDD.map(lambda x: (x, x + 's'))
# Use flatMap
singularAndPluralWordsRDD = wordsRDD.flatMap(lambda x: (x, x + 's'))

# View the results
print singularAndPluralWordsRDDMap.collect()
print singularAndPluralWordsRDD.collect()
# View the number of elements in the RDD
print singularAndPluralWordsRDDMap.count()
print singularAndPluralWordsRDD.count()

[('cat', 'cats'), ('elephant', 'elephants'), ('rat', 'rats'), ('rat', 'rats'), ('cat', 'cats')]
['cat', 'cats', 'elephant', 'elephants', 'rat', 'rats', 'rat', 'rats', 'cat', 'cats']
5
10

Известно, что трансформации groupByKey следует избегать, а то сеть ляжет :)

Let's investigate the additional transformations: groupByKey() and reduceByKey().

Both of these transformations operate on pair RDDs. A pair RDD is an RDD where each element is a pair tuple (key, value). For example, sc.parallelize([('a', 1), ('a', 2), ('b', 1)]) would create a pair RDD where the keys are 'a', 'a', 'b' and the values are 1, 2, 1.

…

Here are more transformations to prefer over groupByKey():

• combineByKey() can be used when you are combining elements but your return type differs from your input value type.
• foldByKey() merges the values for each key using an associative function and a neutral "zero value"

pairRDD = sc.parallelize([('a', 1), ('a', 2), ('b', 1)])
# mapValues only used to improve format for printing
print pairRDD.groupByKey().mapValues(lambda x: list(x)).collect()

# Different ways to sum by key
print pairRDD.groupByKey().map(lambda (k, v): (k, sum(v))).collect()
# Using mapValues, which is recommended when they key doesn't change
print pairRDD.groupByKey().mapValues(lambda x: sum(x)).collect()
# reduceByKey is more efficient / scalable
print pairRDD.reduceByKey(add).collect()

[('a', [1, 2]), ('b', [1])]
[('a', 3), ('b', 1)]
[('a', 3), ('b', 1)]
[('a', 3), ('b', 1)]

Для чего нужны эти трансформации, я так и не понял

Let's investigate the advanced transformations: mapPartitions() and mapPartitionsWithIndex()

# mapPartitions takes a function that takes an iterator and returns an iterator
print wordsRDD.collect()
itemsRDD = wordsRDD.mapPartitions(lambda iterator: [','.join(iterator)])
print itemsRDD.collect()

['cat', 'elephant', 'rat', 'rat', 'cat']
['cat', 'elephant', 'rat', 'rat,cat']

itemsByPartRDD = wordsRDD.mapPartitionsWithIndex(lambda index, iterator: [(index, list(iterator))])
# We can see that three of the (partitions) workers have one element and the fourth worker has two
# elements, although things may not bode well for the rat...
print itemsByPartRDD.collect()
# Rerun without returning a list (acts more like flatMap)
itemsByPartRDD = wordsRDD.mapPartitionsWithIndex(lambda index, iterator: (index, list(iterator)))
print itemsByPartRDD.collect()

[(0, ['cat']), (1, ['elephant']), (2, ['rat']), (3, ['rat', 'cat'])]
[0, ['cat'], 1, ['elephant'], 2, ['rat'], 3, ['rat', 'cat']]

Кэширование RDD

Тут все просто. Если нужно повторно использовать RDD, его надо закешировать.

# Name the RDD
filteredRDD.setName('My Filtered RDD')
# Cache the RDD
filteredRDD.cache()
# Is it cached
print filteredRDD.is_cached
True

# Note that toDebugString also provides storage information
print filteredRDD.toDebugString()

(8) My Filtered RDD PythonRDD[79] at collect at <ipython-input-74-2e6525e1a0c2>:23 [Memory Serialized 1x Replicated]  |  ParallelCollectionRDD[73] at parallelize at PythonRDD.scala:392 [Memory Serialized 1x Replicated]

# If we are done with the RDD we can unpersist it so that its memory can be reclaimed
filteredRDD.unpersist()
# Storage level for a non cached RDD
print filteredRDD.getStorageLevel()
filteredRDD.cache()
# Storage level for a cached RDD
print filteredRDD.getStorageLevel()

Serialized 1x Replicated
Memory Serialized 1x Replicated

Вот, пробежавшись вкратце по основным приемам работы с RDD в Spark, мы можем сделать что-то похожее на реальную работу.

Аппликуха подсчета слов в произведениях Шекспира

 In this lab, we will write code that calculates the most common words in the Complete Works of William Shakespeareretrieved from Project Gutenberg.

Создадим RDD из файла

shakespeareRDD = (sc
                  .textFile(fileName, 8)
                  .map(removePunctuation))
print '\n'.join(shakespeareRDD
                .zipWithIndex()  # to (line, lineNum)
                .map(lambda (l, num): '{0}: {1}'.format(num, l))  # to 'lineNum: line'
                .take(15))

0: 1609
1: 
2: the sonnets
3: 
4: by william shakespeare
5: 
6: 
7: 
8: 1
9: from fairest creatures we desire increase
10: that thereby beautys rose might never die
11: but as the riper should by time decease
12: his tender heir might bear his memory
13: but thou contracted to thine own bright eyes
14: feedst thy lights flame with selfsubstantial fuel

Ой, нам нужна функция removePunctuation

import re
def removePunctuation(text):
    """Removes punctuation, changes to lower case, and strips leading and trailing spaces.
​
    Note:
        Only spaces, letters, and numbers should be retained.  Other characters should should be
        eliminated (e.g. it's becomes its).  Leading and trailing spaces should be removed after
        punctuation is removed.
​
    Args:
        text (str): A string.
​
    Returns:
        str: The cleaned up string.
    """
    res = не буду же я показывать решение лабы?
    return res
print removePunctuation('Hi, you!')
print removePunctuation(' No under_score!')
print removePunctuation(' *      Remove punctuation then spaces  * ')

hi you
no underscore
remove punctuation then spaces

Создадим RDD содержащий отдельные слова

shakespeareWordsRDD = shakespeareRDD.flatMap(lambda line: line.split(' '))
shakespeareWordCount = shakespeareWordsRDD.count()
print shakespeareWordsRDD.top(5)
print shakespeareWordCount

[u'zwaggerd', u'zounds', u'zounds', u'zounds', u'zounds']
927631

Уберем пустые элементы

shakeWordsRDD = shakespeareWordsRDD.filter(lambda x: x)
shakeWordCount = shakeWordsRDD.count()
print shakeWordCount

882996

А теперь посчитаем наиболее употребляемые Шекспиром слова

top15WordsAndCounts = wordCount(shakeWordsRDD).takeOrdered(15, key=lambda (w, cnt): -cnt)
print '\n'.join(map(lambda (w, c): '{0}: {1}'.format(w, c), top15WordsAndCounts))

the: 27361
and: 26028
i: 20681
to: 19150
of: 17463
a: 14593
you: 13615
my: 12481
in: 10956
that: 10890
is: 9134
not: 8497
with: 7771
me: 7769
it: 7678

Ой, нужна функция wordCount

def wordCount(wordListRDD):
    """Creates a pair RDD with word counts from an RDD of words.
​
    Args:
        wordListRDD (RDD of str): An RDD consisting of words.
​
    Returns:
        RDD of (str, int): An RDD consisting of (word, count) tuples.
    """
    res = ну не буду я подсказывать студням.
    return res
print wordCount(wordsRDD).collect()
[('rat', 2), ('elephant', 1), ('cat', 2)]

Вот, собственно, и вся лабораторка.

Поработали с трансформациями и действиями, посчитали сколько раз то или иное слово встречается в тексте.

Освоились, короче.

original post http://vasnake.blogspot.com/2015/08/week-2-lab.html

Week 2

Курс Scalable Machine Learning.

В прошлый раз я рассказал о том, что было на первой неделе – вводная часть.

Теперь материал второй недели – Introduction to Apache Spark.

Topics:

• Big data and hardware trends,
• history of Apache Spark,
• Spark's Resilient Distributed Datasets (RDDs),
• transformations, and actions.

Lab 2: Learning Apache Spark.

... you will learn about the Spark data model, transformations, and actions, and write a word counting program to count the words in all of Shakespeare's plays.

BigData.

Начали с того, что рассказали нам (а то мы не знаем) как количество данных растет такими темпами, что сохранять не успеваем, не то что обрабатывать.

Единственный доступный способ справиться с таким потоком данных – использовать кластеры, где хранение и обработка кусочков данных ведется на узлах сети. Каждый узел – относительно недорогая машинка.

Кстати, никто не почесался задать вопрос: а как оптимизировать отношение количество узлов кластера/мощность отдельного узла.

Хорошо, кластер. Но как справляться с отказами узлов, слабой пропускной способностью сети и, в общем случае, с различающимися характеристиками узлов? Ответ: создать фреймворк, берущий на себя все эти заботы. Сеть это компьютер.

MapReduce.

Идея в том, чтобы разделить датасет на несколько частей, каждую часть обработать на отдельной машине, на этой же машине собрать частичный результат и финальную сборку провести на машине инициировавшей работу.

Если какой-то узел отвалился, его задача перезапускается на другом узле, всего делов-то.

Следовательно, таски MapReduce должны быть идемпотентны и без побочных эффектов (вспоминаем основные особенности функционального программирования).

Hadoop MapReduce, как нам сказали, отличается тем, что для каждой задачи на каждом узле данные считываются и выводятся через диск – долговременную память.

А вот Spark победил эту проблему (проблема с итеративными задачами, частыми в ML – постоянный I/O с диском = тормоза) тем, что датасеты держит в оперативке. И это на некоторых задачах дает возможность получать результат в сто раз быстрее.

Apache Spark.

Короче, в двух словах, как работает Spark:

Берем датасет, разделяем его на коллекции объектов, распределенные по сети, получаем RDD.

Пишем программы обработки в терминах операций над распределенными данными.

Эти операции манипулируют RDD и состоят из трансформаций (превращение одного датасета в другой) и действий (финальная сборка результата).

RDD автомагически восстанавливаются при сбоях узлов.

Больше деталей.

Программа для Spark состоит из двух частей: driver и workers. Драйвер запускает задачу и собирает результаты. Воркеры обрабатывают части датасета на узлах.

Все крутится вокруг RDD и SparkContext. Контекст определяет конфигурацию кластера, что-то вроде коннекта к БД.

RDD – это основной примитив. Датасет иммутабельный, отслеживаемый с целью восстановления при сбое узлов, параллелизуемый.

РДД создается из коллекций (в нашем случе списки Python), трансформацией других РДД или загрузкой файлов с данными из HDFS или из других хранилищ.

Трансформации РДД ленивы. Они не выполняются до тех пор, пока действие (акция) не потребует вычисленных данных.

Spark может выполнять операции с элементами данных в виде key-value. В нашем случае (Python) это tuple (key, value).

Самые известные трансформации с парами, это

• reduceByKey
• sortByKey
• groupByKey

При этом последняя, groupByKey, самая тупая – приводит к перетасовке всего датасета так, чтобы на воркерах оказались записи с одинаковыми ключами. Грубо говоря, через сеть будет пропущен весь датасет. А это очень долго – терабайты и петабайты данных.

Коммуникации.

Проблема: как передавать данные данные между узлами, помимо датасетов RDD?

PySpark создает для каждого воркера, для каждой задачи closure. Это функции, выполняемые над RDD и используемые в функциях глобальные переменные. Но воркеры не могут писать в глобальные переменные. Опять же, глобальные переменные каждый раз засылаются на воркеры, дорого делать их большими.

Решение проблемы. Есть shared variables. Broadcast and Accumulators.

Бродкасты предназначены для отсылки на воркеров больших объемов данных. Только чтение.

Аккумуляторы нужны для сбора данных на драйвере от воркеров.

Что-то много выходит. Про лабораторку напишу в следующий раз.

original post http://vasnake.blogspot.com/2015/08/week-2.html

Week 1

Курс Scalable Machine Learning.

В прошлый раз я рассказал о том, как мы делали лабораторки. Теперь материал первой недели.

Topics:

• Apache Spark overview,
• basic machine learning concepts,
• steps of typical supervised learning pipelines,
• linear algebra review,
• computational complexity / big O notation review.

Lab 1:

NumPy, Linear Algebra, and Lambda Function Review.

Gain hands on experience using Python's scientific computing library to manipulate matrices and vectors, and learn about lambda functions which will be used throughout the course.

Apache Spark это такая кластерная вычислялка, основная идея которой заключается в том, что исходный набор данных оборачивается в RDD – Resilient Distributed Dataset, который кусками раскидывается на по вычислительным нодам и эти куски там параллельно обрабатываются (стадия Map). На стадии Reduce результаты вычислений возвращаются на ноду-driver, инициировавшую обработку. Особенно отмечается тот факт, что, в отличие от традиционного MapReduce, в Spark данные крутятся в оперативной памяти, что позволяет достичь стократного превосходства по быстродействию.

Дальше – machine learning, концепция.

Галопом-по-европам, огромные датасеты, их надо очистить, преобразовать, извлечь фичи/свойства, нормализовать, найти по ним оптимум для разных моделей, оценить модели, выбрать лучшую и, в светлом будущем, использовать модель для предсказаний.

Линейная алгебра.

Векторы, матрицы, inner product, outer product, разные свойства операций над матрицами. В общем, там где сумма произведений – там ищи применение умножению матриц.

А еще транспонированные матрицы, инвертированные матрицы, identity matrix, Euclidean norm for vectors, вот это все.

В наших вычислениях (ML) используется повсеместно и очень плотно.

Big O notation.

Довольно важная штука. Чтобы оценить, потянет машины или кластер обработку набора данных, определенного размера, надо уметь оценить предполагаемые затраты на хранение и вычисления. Для используемого алгоритма. Разные алгоритмы имеют разные оценки по времени/размеру.

Время оценивается в количестве элементарных операций, размер оценивается в единицах хранения – 8 байт под плавующую точку.

А потом была лабораторка.

https://github.com/spark-mooc/mooc-setup/blob/master/ML_lab1_review_student.ipynb

Сначала на бумажке вычисляли результаты умножения матриц, записывая результаты в нотебук.

Потом умножали матрицы по всякому с помощью NumPy.

Посмотрели на DenseVector.

Потом занялись lambda и имитацией map().reduce() под соусом «функциональный подход».

На этом первая неделя как-то закончилась.

Ничего серьезного.

original post http://vasnake.blogspot.com/2015/08/week-1.html