Amazon Kinesis

# Criar stream
aws kinesis create-stream --stream-name my-stream --shard-count 2

# Verificar status
aws kinesis describe-stream --stream-name my-stream

# Listar streams
aws kinesis list-streams

import boto3
import json
import time
from datetime import datetime

# Criar cliente Kinesis
kinesis = boto3.client('kinesis', region_name='us-east-1')

def put_record(stream_name, data, partition_key):
    try:
        response = kinesis.put_record(
            StreamName=stream_name,
            Data=json.dumps(data),
            PartitionKey=partition_key
        )
        print(f"Record sent: {response['SequenceNumber']}")
        return response
    except Exception as e:
        print(f"Error sending record: {e}")

# Exemplo de uso
for i in range(100):
    data = {
        'timestamp': datetime.now().isoformat(),
        'user_id': f'user_{i}',
        'event': 'page_view',
        'page': f'/page_{i % 10}'
    }
    
    put_record('my-stream', data, f'user_{i}')
    time.sleep(0.1)

import boto3
import json
import time

kinesis = boto3.client('kinesis', region_name='us-east-1')

def get_shard_iterator(stream_name, shard_id):
    response = kinesis.get_shard_iterator(
        StreamName=stream_name,
        ShardId=shard_id,
        ShardIteratorType='LATEST'
    )
    return response['ShardIterator']

def consume_records(stream_name):
    # Obter informações do stream
    stream_info = kinesis.describe_stream(StreamName=stream_name)
    shards = stream_info['StreamDescription']['Shards']
    
    # Processar cada shard
    for shard in shards:
        shard_id = shard['ShardId']
        shard_iterator = get_shard_iterator(stream_name, shard_id)
        
        while shard_iterator:
            try:
                response = kinesis.get_records(ShardIterator=shard_iterator)
                records = response['Records']
                
                for record in records:
                    data = json.loads(record['Data'])
                    print(f"Received: {data}")
                
                shard_iterator = response.get('NextShardIterator')
                time.sleep(1)
                
            except Exception as e:
                print(f"Error reading records: {e}")
                break

# Consumir records
consume_records('my-stream')

-- Criar stream de entrada
CREATE STREAM "SOURCE_SQL_STREAM_001" (
    "timestamp" VARCHAR(32),
    "user_id" VARCHAR(32),
    "event" VARCHAR(32),
    "page" VARCHAR(64)
);

-- Análise em tempo real - contagem de eventos por página
CREATE STREAM "DESTINATION_SQL_STREAM" AS 
SELECT 
    "page",
    COUNT(*) as "event_count",
    ROWTIME_TO_TIMESTAMP(ROWTIME) as "window_time"
FROM "SOURCE_SQL_STREAM_001"
GROUP BY "page", 
         ROWTIME RANGE INTERVAL '1' MINUTE;

-- Detecção de anomalias - usuários com muitos eventos
CREATE STREAM "ANOMALY_STREAM" AS
SELECT 
    "user_id",
    COUNT(*) as "event_count",
    ROWTIME_TO_TIMESTAMP(ROWTIME) as "window_time"
FROM "SOURCE_SQL_STREAM_001"
GROUP BY "user_id",
         ROWTIME RANGE INTERVAL '5' MINUTE
HAVING COUNT(*) > 100;

# Kinesis Stream
resource "aws_kinesis_stream" "example" {
  name             = "my-stream"
  shard_count      = 2
  retention_period = 48

  shard_level_metrics = [
    "IncomingRecords",
    "OutgoingRecords",
  ]

  tags = {
    Environment = "production"
    Application = "streaming-app"
  }
}

# Kinesis Firehose
resource "aws_kinesis_firehose_delivery_stream" "example" {
  name        = "my-firehose"
  destination = "s3"

  s3_configuration {
    role_arn   = aws_iam_role.firehose_role.arn
    bucket_arn = aws_s3_bucket.bucket.arn
    prefix     = "data/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
    
    buffer_size     = 5
    buffer_interval = 300
    
    compression_format = "GZIP"
  }
}