В этой статье описывается эксперимент, который я пробовал с AWS Lambda, Kinesis Streams и Serverless Framework. Я хотел увидеть, насколько легко я могу запустить код для решения одной проблемы на сотнях серверов параллельно, используя бессерверные сервисы AWS, управляемые событиями. Я хотел использовать Serverless Framework для создания чего-то простого, что можно было бы очень быстро развернуть, запустить и снова урезать.

Раньше я пытался решить загадку «Завершение королевы» и пришел к следующему:



Решение головоломки завершения N-Queens - Джеймс Робинсон - средний
Эта статья представляет собой предысторию некоторого кода, который я использовал для этой статьи об использовании сервисов AWS для запуска параллельного кода… medium.com



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

Код был написан на:

  • взять запись из очереди
  • обработать его, и в результате…
  • создать еще несколько записей для обработки или
  • вывести решение проблемы

Если мы поместим эти записи во внешнюю очередь, у нас может быть несколько рабочих (лямбда-функций), обрабатывающих эти записи и распределяющих рабочую нагрузку по многим процессорам. Сможет ли он найти приличное время все решения шахматной доски 28 x 28, которая пока остается нерешенной?

Спойлеры

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

Бессерверная структура

Так что, если вы не знаете, что такое Serverless Framework, среди прочего, она может упростить процесс развертывания кода в виде функций Lambda на AWS (или аналогичных в Azure / GAE и т. Д.) И настройки триггеров для те функции, которые будут вызываться. См. Другую мою статью для объяснения того, как развернуть с ним веб-службу REST:



Как создать простой веб-сервис REST с помощью Node, AWS Lambda и Serverless Framework
В этой статье я объясню, как создать веб-сервис REST, который написан с Node.js и развернут в AWS Lambda… medium.com



Kinesis Streams

Kinesis Streams - это в основном очереди, но одна вещь, которую они предлагают через SQS (простая служба очередей Amazon), - это то, что они могут быть разделены на несколько сегментов. Сообщения, помещенные в начало очереди, могут равномерно распределяться по этим осколкам. Затем вы можете присоединить потребителей к этим осколкам / подочередям. Таким образом, у вас может быть один `` поток '', но вы можете предоставить ему до 500 шардов и распределять рабочую нагрузку параллельно между 500 потребителями, считывая со скоростью 5 чтений в секунду с каждого шарда, с максимальной скоростью чтения 2 МБ / с. . С вас взимается плата за время развертывания шардов из расчета 0,0179 доллара США за сегмент в час.

На этом этапе вы также не можете подключить лямбда-функцию к очереди SQS, поэтому ... Kinesis выигрывает.

Я буду хранить все решения, которые предлагает моя система, в таблице DynamoDB, поэтому мы получим что-то вроде этого:

Kinesis + Lambda в Serverless.yml

Вот как вы можете создать поток Kinesis и присоединить к нему лямбда-функцию с помощью Serverless. Я объясню свой файл serverless.yml. (Вы можете найти все здесь)

service: queens-completion-lambda-kinesis 
provider: 
  name: aws 
  runtime: nodejs6.10 
  region: eu-west-2 
  memorySize: 128 
  timeout: 300

Это сообщает Serverless, что мы хотим использовать AWS, Node v6.10 и развернуть в регионе EU-West-2, нам нужны экземпляры Lambda со 128 МБ ОЗУ, и они должны отключиться через 300 секунд (максимально допустимое время). MemorySize также определяет, какой процессор мы собираемся получить. AWS Lambda распределяет мощность процессора пропорционально памяти (см. Настройка лямбда-функций), поэтому, хотя нам может не понадобиться много памяти, мы можем получить лучшие результаты позже, если увеличим ее.

Привилегии

iamRoleStatements: # Grant privilege read from Kinesis work stream 
  - Effect: Allow 
    Action: 
      - kinesis:GetRecords 
      - kinesis:GetShardIterator 
      - kinesis:DescribeStream 
      - kinesis:ListStreams 
      - kinesis:PutRecord 
      - kinesis:PutRecords 
    Resource: 
      Fn::GetAtt: 
        - workStream  
        - Arn

У нас будет поток Kinesis под названием «workStream». Чтобы наши функции Lambda могли читать / писать в этот поток с помощью AWS SDK, нам необходимо знать ARN (имя ресурса Amazon) для этой таблицы. Мы не узнаем этого, пока таблица не будет создана, и она не будет создана в разделе ресурсов этой бессерверной конфигурации, поэтому мы используем функцию GetAtt (получить атрибут), чтобы получить ее.

# Grant privilege to write to results table 
- Effect: Allow
  Action:
  - dynamodb:PutItem
  Resource:
    Fn::GetAtt: 
    - resultsTable
    - Arn

Для тестирования я также хотел создать таблицу «результатов» DynamoDB, чтобы улавливать решения, которые она предлагает. Эта конфигурация дает ему право записи в таблицу результатов, которую я создам позже.

Лямбда-функция

functions:
  queensHandler:
    handler: lambdaHandler.process

Это моя лямбда-функция. Все, что я здесь сделал, это объявил новую лямбду с именем queensHandler и установил функцию process (), экспортированную lambdaHandler.js, в качестве кода для этого. К сожалению, я не могу подключить его к потоку Kinesis здесь, используя бессерверную структуру, поэтому мне нужно сделать это с помощью некоторой прямой конфигурации CloudFormation ниже.

Kinesis stream

resources:
  Resources:
    workStream: 
      Type: AWS::Kinesis::Stream
      Properties: 
        Name: queensWorkStream
        ShardCount: 200

В разделе ресурсов файла serverless.yml мы можем включить некоторые вещи CloudFormation для создания нашего потока Kinesis, таблицы результатов и сопоставления потока с нашей лямбда-функцией. Во-первых, у нас есть поток. Все очень просто. Это количество сегментов определит, сколько лямбд мы будем слушать для работы. Он не масштабируется автоматически, поэтому вам придется делать это самостоятельно, если вы хотите, чтобы он масштабировался в соответствии с рабочей нагрузкой или упреждающе на периоды занятости.

Таблица DynamoDB

resultsTable:
  Type: AWS::DynamoDB::Table
  Properties:
    TableName: queensResultsTable
    AttributeDefinitions:
      - AttributeName: key
        AttributeType: S
    KeySchema:
      - AttributeName: key
        KeyType: HASH
    ProvisionedThroughput:
      ReadCapacityUnits: 1
      WriteCapacityUnits: 100

Здесь я создал таблицу результатов. Мы не будем читать из него, поэтому я установил емкость чтения как минимум 1, но у нас будет много лямбда-выражений, записывающих в него одновременно.

Кинезис - ›Лямбда-отображение.

Event:
  Type: AWS::Lambda::EventSourceMapping
  Properties:
    BatchSize: 50
    EventSourceArn:
      Fn::GetAtt:
        - workStream
        - Arn      
    FunctionName:
      Fn::GetAtt:
        - QueensHandlerLambdaFunction
        - Arn
    StartingPosition: TRIM_HORIZON

Вот где я фактически прикрепляю к потоку нашу новую лямбда-функцию. BatchSize сообщает, сколько элементов нужно собрать и передать функции Lambda. Хотя, если вы поместите в осколок только 10 элементов, очевидно, что он не будет ждать вечно, пока он не наберется 20. Если вы хотите узнать больше об этом механизме, есть отличная статья Алессандро Болоньи https://itnext.io/creating-a -blueprint-for-microservices-and-event-sourcing-on-aws-291d4d5a5817

Обратите внимание, что я должен указать имя Lambda как: QueensSolverLambdaFunction в вызове функции GetAtt (получить атрибут), несмотря на то, что я назвал его «queensSolver» выше. ¯ \ _ (ツ) _ / ¯

Код

Итак, это развертывание. Что касается кода, у меня есть моя функция, выполняющая работу, которая выглядит следующим образом: processSet(set, onNewSet, onResult) Она принимает некоторые данные из очереди и выполняет обратные вызовы, чтобы предоставить другой набор / запись для помещения в очередь / поток или предоставить Результат он нашел, что мы можем сохранить в таблице результатов. Он ничего не знает о реализации очереди. Затем у нас есть наш lambdaHandler.js файл, который принимает данные из очереди и преобразует их для передачи:

function process(event, context, callback) {
  event.Records.forEach(record => {
    const set = convert(record)
    // Pass data and callback functions to processor 
    processSet(set, addNewSet, publishResult)
  })
}
module.exports.process = process

Он также содержит две функции, которые будут принимать выходные данные из функции и либо помещать эти выходные данные в очередь, либо сохранять их в таблице результатов. Вот код для добавления новой записи в наш поток Kinesis:

function addNewSet(set) { 
  const key = set.key.toString(16) 
  const params = { 
    Data: JSON.stringify({ 
      level: set.level, 
      key: key, 
      connections: set.connections.toString(16), 
    }), 
    PartitionKey: key, 
    StreamName: config.KINESIS_WORK_STREAM, 
  } 
  // Write record to stream 
  kinesis.putRecord(params, err => { 
    if(err) console.error(err, err.stack) 
  }) 
}

AWS закодирует данные в формате base64 за вас. Ключ разделения - это значение, которое AWS будет пропускать через хэш-функцию, чтобы определить, на каком сегменте будут находиться данные, поэтому, если вы хотите, чтобы один процессор сегментов получал все записи определенного типа-кода (для целей кеширования / производительности ), то вы можете использовать этот типовой код в качестве ключа раздела. В противном случае используйте что-то уникальное, чтобы как можно более равномерно распределить записи по всем шардам.

Вот наша функция для преобразования записей из очереди обратно в объекты для нашей функции процесса:

function convert(record) { 
  // Convert from base64 to Buffer to JSON string 
  const jsonStr =
    Buffer.from(record.kinesis.data,'base64').toString() 
  // ...to an object 
  const data = JSON.parse(jsonStr) 
  // ...to a set we can process. 
  // Each set has a key (a map of nodes), a level (number of nodes 
  // in set) and 'connections' (map of mutually connected nodes). 
  // Convert from hex string to big number 
  const set = { 
    level: data.level, 
    key: new BN(data.key, 16), 
    connections: new BN(data.connections, 16) 
  } 
  return set 
}

Весь код для этого можно найти здесь:



bishbashbosh123 / queens-completion-lambda-kinesis
queens-completion-lambda-kinesis - решатель головоломки с завершением кинезиса / лямбда-ферзей github.com



Полученные результаты

Хорошо, давайте запустим ...

$ serverless deploy

Я создал сценарий под названием putInitialRecord.js, чтобы поместить запись с ключом 0 в мой поток. Это побуждает мою функцию processSet поместить исходные записи для обработки в очередь; В контексте этой проблемы одна запись на каждое поле на шахматной доске.

$ node putInitialRecord

После того, как вы закончите, очистите все, чтобы избежать ненужных затрат:

$ serverless remove

Размер платы: 8, осколки: 50, размер пакета: 50, память: 128 МБ
92. решение найдено за: 231 секунду

Учитывая, что на моем локальном компьютере это занимает 0,4 секунды, это довольно досадно, но эй! Я просто запустил код сразу на 100 серверах, и это обошлось мне примерно в два цента (0,0179 доллара за шард в час). Через несколько секунд после включения я вижу, что имеется 50 серверов, поскольку в Cloudwatch имеется 50 потоков журналов - ›Журналы, в каждом из которых указывается время начала / окончания и продолжительность каждого выполнения Lambda.

Я попробую изменить эти параметры:

Размер платы: 8, осколки: 50, размер пакета: 100, память: 128 МБ
92. решение найдено за: 223 секунды

Размер платы: 8, осколки: 50, размер пакета: 50, память: 256 МБ
92. решение найдено за: 71 секунду

Размер платы: 8, осколки: 100, размер пакета: 50, память: 128 МБ
92 решения найдены за: 70 секунд

Приятно видеть, что в два раза больше памяти (и, следовательно, в два раза больше мощности процессора) обрабатывается за то же время, что и в два раза больше рабочих. Даже в этом случае он все равно не превзойдет 0,4 секунды. Но, может быть, если бы проблема была больше, лучше было бы делать все это параллельно.

Еще один тест

Размер платы: 10, осколки: 200, размер пакета: 50, память: 256 МБ
724. решения, найденные за: 19 минут 54 секунды

Локально это занимает 10 секунд. Хорошо, я определенно не собираюсь пробовать что-либо близкое к доске 28 x 28. Но все же ... Я впечатлен тем, как легко раскрутить что-то подобное.

Резюме

Так что я не получил хороших результатов для своего конкретного эксперимента, но удивительно, насколько легко, дешево и быстро можно развернуть такое количество серверов для того, что вы хотите делать. Я использовал экземпляры Lambda базового уровня, но вы можете получить гораздо больше мощности, если готовы за это заплатить.

Kinesis Streams - отличный и простой сервис для такого рода горизонтального масштабирования. Мои единственные затраты составили всего 0,0179 доллара в час за сегмент.
100 из них на полчаса = 0,90 доллара США + 0,01 доллара США за миллион запросов на размещение в очереди.
Все остальное (вызовы лямбда-кода, модули чтения / записи DynamoDB) входило в бесплатный уровень. Хотя убедитесь, что вы выключили систему (serverless remove) после того, как закончите играть ... эти модули DynamoDB и осколки Kinesis не уменьшаются, когда вы их не используете.
100 единиц записи на единицу по цене 0,0008 долларов США на запись на единицу в час =
58 долларов в месяц.

Бессерверная структура упрощает возможности CloudFormation; определение архитектуры вашего приложения в одном «стеке», которое будет развернуто по вашему выбору, и его абстракция, чтобы упростить развертывание где по вашему выбору (например, в Azure или других поставщиках). Он получает большую часть своей мощности от полного контроля, который может предоставить вам AWS SDK; Если вы хотите написать код для автоматического масштабирования ваших шардов Kinesis или контролировать свои расходы на выставление счетов, вы можете это сделать. Дай мне знать, если ты сделаешь с ним что-нибудь крутое.