Отсутствующие записи при получении из потока данных Kinesis с использованием getRecords - PullRequest
Новая
       8

Отсутствующие записи при получении из потока данных Kinesis с использованием getRecords

0 голосов
/ 14 декабря 2018

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

Я нахожу данные отсутствующими при получении из getRecords () по сравнению с получением данных в S3 с использованиемПожарный пользователь обращается к тому же потоку данных

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

<?php

require 'vendor/autoload.php';

use Aws\Kinesis\KinesisClient;

$streamName = 'DATA-STREAM-NAME';
$kinesisClient = KinesisClient::factory(array(
                                        'region' => 'eu-west-1',
                                        'version' => '2013-12-02'
                                            )
                                    );

// Get shard list
$shardList = $kinesisClient->listShards([
    'StreamName' => $streamName,
]);

// Iterate through shards to get iterators
$iterators = array();
foreach ($shardList['Shards'] as $shard) {
    $shardIterator = $kinesisClient->getShardIterator([
        'ShardId' => $shard['ShardId'],
        'ShardIteratorType' => 'LATEST',
        'StreamName' => $streamName
    ]);
    $iterators[$shard['ShardId']] = $shardIterator['ShardIterator'];
}

// Loop continuously to get feed from kinesis
while (1) {
    $iterators = getLogs($iterators);
    sleep(3);
}

// Get records from each iterator
function getLogs($iterators) {
    global $kinesisClient;
    $nextIterators = array();
    foreach ($iterators as $key => $iterator) {
        $records = $kinesisClient->getRecords(['ShardIterator' => $iterator]);
        $nextIterators[$key] = $records['NextShardIterator'];
        if (empty($records['Records'])) {
            return $nextIterators;
        }
        foreach ($records['Records'] as $data) {
            echo $data['Data'];
        }
    }
    return $nextIterators;

}
...