Как использовать ExplicitHashKey для назначения потоков с циклическим перебором в AWS Kinesis

Я пытаюсь перекачивать много данных через Amazon Kinesis (порядка 10 000 точек в секунду).

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

Казалось бы, я мог бы сделать это с параметром ExplicitHashKey для сообщений в списке, который я отправляю в конечную точку PutRecords, однако документация Amazon фактически не описывает, как использовать ExplicitHashKey, кроме заявления оракула:

http://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html

Каждая запись в массиве Records может включать необязательный параметр ExplicitHashKey, который переопределяет ключ раздела для сопоставления сегментов. Этот параметр позволяет производителю данных явно определять сегмент, в котором хранится запись. Дополнительную информацию см. В разделе Добавление нескольких записей с помощью PutRecords в Руководстве разработчика Amazon Kinesis Streams.

(В заявлении в документации выше есть ссылка на другой раздел документации, в котором вообще не обсуждается ExplicitHashKeys).

Есть ли способ использовать ExplicitHashKey для циклического перебора данных между шардами?

Какие допустимые значения параметра?


amazon-kinesis amazon-kcl
person deadcode    schedule 16.06.2017    source источник
comment
Как ExplicitHashKey минимизировать количество разделов? Разве вы не хотите предоставить хороший partitionKey, для которого kinesis сгенерирует MD5 и решит, в какой раздел он будет помещен? Допустим, у вас 2 раздела, и вы выбрали хороший partitionKey. Затем Kinesis partitioner отправит эту запись в один из этих двух разделов на основе 128-битного значения.   -  person prayagupd    schedule 16.06.2017
comment
@prayagupd - Предположим, вы хотите обрабатывать 9900 точек данных в секунду. Поскольку Kinesis поддерживает 1000 точек в секунду на каждый сегмент, вы можете использовать 10 сегментов. Однако, если распределение ваших данных по осколкам не является идеальным, рано или поздно один осколок будет получать более 1000 баллов в секунду. В качестве практического примера у меня есть система, через которую я пропускаю 13 000 точек в секунду и использую случайное число в качестве ключа раздела, и даже с 23 шардами я все еще вижу исключения из предела скорости в редких случаях. Если бы у меня было идеальное круговое распределение, я мог бы использовать 13.   -  person deadcode    schedule 17.06.2017
comment
Попался. Но как вы тогда относитесь к круговой системе? Я мог бы подумать о том, чтобы иметь некоторый порядковый номер как partitionKey, чтобы он равномерно переходил в sequentialPartitionKey / np. Для простоты предположим, что один раздел может записывать до 10 событий / 1000 мс. Для скорости 130 событий / 1000 мс, скажем, порядковые номера от 1 до 130, тогда будет ровно 10 событий 1, 14, 27, 40, 53, 66, 79, 92, 105, 118, идущих в раздел1, независимо от того, в каком порядке они обрабатываются, то же самое и с другими разделами. Похоже, вы на правильном пути, устанавливая свой собственный ключ раздела вместо хеша MD5.   -  person prayagupd    schedule 17.06.2017
comment
Теперь, когда я лучше понимаю ситуацию, в моей конкретной ситуации я могу выполнять циклический перебор, отправляя каждую N-ю запись в str (int (((N% NUM_SHARDS) +0.5) * (2 ** 128 / NUM_SHARDS))). Это потому, что все мои шарды имеют диапазоны хэш-ключей одинакового размера. Вместо этого вы можете описать поток, определить диапазоны хэш-ключей динамически и выбрать число в каждом диапазоне, а затем выполнить циклический перебор этих чисел.   -  person deadcode    schedule 17.06.2017
comment
Хотя я не понял, str( int( (( N%NUM_SHARDS )+0.5) * ( 2**128 / NUM_SHARDS ) ) ) выглядит интересно.   -  person prayagupd    schedule 29.06.2017

Ответы (1)


arrow_upward
9
arrow_downward

Каждому сегменту назначается последовательный диапазон 128-битных целых чисел от 0 до 2 ^ 128-1.

Вы можете найти диапазон целых чисел, назначенных данному сегменту в потоке, через интерфейс командной строки AWS:

aws kinesis describe-stream --stream-name name-of-your-stream

Результат будет выглядеть так:

{
    "StreamDescription": {
        "RetentionPeriodHours": 24, 
        "StreamStatus": "ACTIVE", 
        "StreamName": "name-of-your-stream", 
        "StreamARN": "arn:aws:kinesis:us-west-2:your-stream-info", 
        "Shards": [
           {
                "ShardId": "shardId-000000000113", 
                "HashKeyRange": {
                    "EndingHashKey": "14794885518301672324494548149207313541", 
                    "StartingHashKey": "0"
                }, 
                "ParentShardId": "shardId-000000000061", 
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "49574208032121771421311268772132530603758174814974510866"
                }
            }, 
           { ... more shards ... }
       ...

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

Обратите внимание, что из-за предыдущих операций слияния и разделения на вашем сегменте может быть много сегментов с перекрытием HashKeyRanges. В настоящее время открытые осколки не имеют элемента SequenceNumberRange.EndingSequenceNumber.

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

В качестве примечания, вы также можете вычислить хеш-значение, которое будет оценивать данный PartitionKey, следующим образом:

  1. Вычислите сумму MD5 ключа раздела.
  2. Интерпретируйте сумму MD5 как шестнадцатеричное число и преобразуйте ее в основание 10. Это будет хэш-ключ для этого ключа раздела. Затем вы можете посмотреть, в какой осколок попадает этот хеш-ключ.
person deadcode    schedule 16.06.2017