kinesis-dump-shard.rb

require 'logger'
require 'aws-sdk'

log = Logger.new($stderr)
log.level = Logger::DEBUG

stream_name = ARGV[0]

# http://docs.aws.amazon.com/kinesis/latest/dev/kinesis-using-iam.html

AWS.config({
  access_key_id:     ENV['DEVELOPMENT_KINESIS_KEY_ID'],
  secret_access_key: ENV['DEVELOPMENT_KINESIS_ACCESS_KEY']
})

stream = AWS::Kinesis::Client.new.describe_stream(
  stream_name: stream_name
)

log.info "stream.stream_description.inspect: #{stream.stream_description.inspect}"
log.info "stream.stream_description.shards.size: #{stream.stream_description.shards.size}"

shard = stream.stream_description.shards.first

k = AWS::Kinesis::Client.new
# If a GetShardIterator request is made too often, you receive a
# ProvisionedThroughputExceededException . For more information about throughput
# limits, see GetRecords.
iterator = k.get_shard_iterator(
  stream_name: stream_name,
  shard_id: shard.shard_id,
  shard_iterator_type: 'TRIM_HORIZON',
  # starting_sequence_number: starting_sequence_number
)

next_iterator_id = iterator.shard_iterator

loop do
  log.info "next_iterator_id: #{next_iterator_id}"
  data = k.get_records(
    shard_iterator: next_iterator_id
  )
  next_iterator_id = data[:next_shard_iterator]
  data.records.each do |data|
    log.info data.inspect
  end
  sleep 2
end