Questions tagged with Amazon Kinesis

I have some IoT devices that send raw data which I am planning to store in Kafka and then process it with the consumer and save it to a database. Scenario: IoT devices send sensor data in form of packets (1 packet per sec) and you need 10 raw packets to decode its value (I have a C binary that takes 10 packets, I can enforce in IoT to send 10 packets combined as a single packet but I want IoT to focus on single packet only so that in future I can modify the algorithm to work for 20, 30 packets on the cloud). If I use the same topic for all IoT then consumers have to wait until it has received 10 packets to start processing, which I am thinking can cause some problems. (Problem of filtering packet of IoT, as single packets, can mix up in single topic) Issues: Initially, I thought of creating a single topic with partitions for each IoT device, but Kafka doesn't support dynamic partitions. (More IoT devices can be added or removed during runtime.) Another solution can be to create a topic for each IoT device with 1 partition. Also, the above approach of 1 topic for each IoT device may enforce to use of one consumer for each IoT device added (can this be improved?) I am looking for a better and more efficient implementation if possible for this problem. The number of IoT devices is dynamic and can go up to 100+.

Hi There, I am trying out Kinesis Client in [JS SDK V3](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-kinesis/globals.html). When I create a Kinesis Client in the global scope and reuse the same client for all further Kinesis ingestion, I am noticing that after a while I am getting the following error. ``` 1|new-sig | { Error [ERR_HTTP2_INVALID_SESSION]: The session has been destroyed 1|new-sig | at ClientHttp2Session.request (internal/http2/core.js:1559:13) 1|new-sig | at Promise (/home/ec2-user/signaling-v7.temasys.io/node_modules/@aws-sdk/node-http-handler/dist-cjs/node-http2-handler.js:57:33) 1|new-sig | at new Promise (<anonymous>) 1|new-sig | at NodeHttp2Handler.handle (/home/ec2-user/signaling-v7.temasys.io/node_modules/@aws-sdk/node-http-handler/dist-cjs/node-http2-handler.js:37:16) 1|new-sig | at stack.resolve (/home/ec2-user/signaling-v7.temasys.io/node_modules/@aws-sdk/client-kinesis/dist-cjs/commands/PutRecordCommand.js:27:58) 1|new-sig | at /home/ec2-user/signaling-v7.temasys.io/node_modules/@aws-sdk/middleware-serde/dist-cjs/deserializerMiddleware.js:5:32 1|new-sig | at /home/ec2-user/signaling-v7.temasys.io/node_modules/@aws-sdk/middleware-signing/dist-cjs/middleware.js:11:26 1|new-sig | at process._tickCallback (internal/process/next_tick.js:68:7) '$metadata': { attempts: 1, totalRetryDelay: 0 } } ``` I am not getting an idea as to why this is happening. However when I use a custom requestHandler and disable Keep-Alive the client stops throwing the error (like the below code) ``` const { NodeHttpHandler } = require("@aws-sdk/node-http-handler"); const { Agent } = require("http"); const kinesisClient = new KinesisClient({ region: kinesisDataStream.region, requestHandler: new NodeHttpHandler({ httpAgent: new Agent({keepAlive: false}) })}); ``` Could you help me in understanding what's going on ? Thanks a lot.

I have a Kinesis data Stream whose records I want to insert it in the AWS redshift with using AWS Glue.I created crawlers to bring source table and target table .They are working fine with . The code works when I save the records on S3 instead of Redshift .But When using Target as Redshift I get error. However I keep getting error "StreamingQueryException: Error while List shards"

I've been billed about $30/month for Kinesis "OnDemandStreamHr" at about 720hr/month (so running 24hrs) for the last few months. I was playing around with Kinesis in January when studying for the ML cert, which is when the billing started, so that tracks, but I was very careful to shut down all my instances. I still don't see any running kinesis instances despite the billing continuing to happen. Any insight would be much appreciated.

I want to get email log(ses) by kinesis with s3. So I tried this docs. https://docs.aws.amazon.com/ko_kr/ses/latest/dg/event-publishing-kinesis-analytics-configuration-set.html But Problem happened. I tried "added destination" at Configuration Sets. I also settings like this -> https://docs.aws.amazon.com/ses/latest/dg/event-publishing-add-event-destination-firehose.html#event-publishing-add-event-destination-firehose-role But, the error message like this -> Could not access Kinesis Firehose Stream using IAM role What should I do?

i am trying to consume data using python script from kinesis data stream which is created successfully and data is produced or streamed to it successfully , but when running consumer script in python : ``` import boto3 import json from datetime import datetime import time my_stream_name = 'stream_name' kinesis_client = boto3.client('kinesis', region_name='us-east-1') response = kinesis_client.describe_stream(StreamName=my_stream_name) my_shard_id = response['StreamDescription']['Shards'][0]['ShardId'] shard_iterator = kinesis_client.get_shard_iterator(StreamName=my_stream_name, ShardId=my_shard_id, ShardIteratorType='LATEST') my_shard_iterator = shard_iterator['ShardIterator'] record_response = kinesis_client.get_records(ShardIterator=my_shard_iterator, Limit=2) while 'NextShardIterator' in record_response: record_response = kinesis_client.get_records(ShardIterator=record_response['NextShardIterator'], Limit=2) print(record_response) # wait for 5 seconds time.sleep(5) ``` But the output of the message data is empty ('Records': []): ``` {'Records': [], 'NextShardIterator': 'AAAAAAAAAAFFVFpvvveOquLUe7WO9nZAcYNQdcS6f6a+YGrrrjZo1gULcu/ZYxC7AB+xVlUhgL9UFPrQ22qmcQa6iIsmuKWl26buBk3utXlVqiGuDUYSgqMOtkp0Y7pJwa6N/I0fYfl2PLTXp5Qz8+5ZYuTW1KDt+PeSU3992bwgdOm7744cxcSnYFaQuHqfa0vLlaRBTOACVz4fwjggUBN01WdsoEjKmgtfNmuHSA7s9LLNzAapMg==', 'MillisBehindLatest': 0, 'ResponseMetadata': {'RequestId': 'e451dd27-c867-cf3d-be83-edbe95e9da9f', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'e451dd27-c867-cf3d-be83-edbe95e9da9f', 'x-amz-id-2': 'ClSlC3gRJuEqL9YJcHgC2N/TLSv56o+6406ki2+Zohnfo/erFVMDpPqkEWT+XAeeHXCdhYBbnOeZBPyesbXnVs45KQG78eRU', 'date': 'Thu, 14 Apr 2022 14:23:21 GMT', 'content-type': 'application/x-amz-json-1.1', 'content-length': '308'}, 'RetryAttempts': 0}} ```

When using Amazon Inspector, snyk finds 3 CVEs in images which contain the latest release of the Amazon Kinesis Agent dependencies. Be aware, VULN scanners are dumb and I have no evidence these vulnerabilities are currently exploitable given the way the agent uses those libraries. Furthermore, I have made no effort to confirm that one way or the other. Instead, I simply built a version of the agent which includes later versions of the dependencies which have addressed the known VULNs. One could argue the efficacy of that approach, but my primary intent is to increase the signal to noise ratio of my VULN scans by reducing noise. Unfortunately, while there appear to be a set of unit/integration tests in the github repository, there are no instructions on how to run those tests. AFAICT they have not been updated for a couple years. So a series of questions: 1. Are there instructions anywhere for how to run that test suite, and is there any confidence it still works? 2. Is anyone else interested in helping me test this new build: https://github.com/britive/amazon-kinesis-agent or https://github.com/britive/amazon-kinesis-agent/raw/master/rpm/aws-kinesis-agent-2.0.6-1b.amzn2.noarch.rpm Also see: https://github.com/awslabs/amazon-kinesis-agent/issues/242 Thanks

I followed [this AWS blog post](https://aws.amazon.com/blogs/aws/introducing-amazon-kinesis-data-analytics-studio-quickly-interact-with-streaming-data-using-sql-python-or-scala/) to create KDA app, and change the output sink into s3 instead of data stream, everything is working, and I can get the result in s3. However in the KDA error logs, glue keep throwing getFunction error almost every second I run the deployed app, but I only use glue to define input/output schemas, didn't use any glue function, so I wonder where is it come form. Please help to take a look. ``` @logStream kinesis-analytics-log-stream @message {"locationInformation":"com.amazonaws.glue.catalog.metastore.GlueMetastoreClientDelegate.getFunction(GlueMetastoreClientDelegate.java:1915)","logger":"com.amazonaws.glue.catalog.metastore.GlueMetastoreClientDelegate","message":"software.amazon.kinesisanalytics.shaded.com.amazonaws.services.glue.model.EntityNotFoundException: Cannot find function. (Service: AWSGlue; Status Code: 400; Error Code: EntityNotFoundException; Request ID: <Request ID>; Proxy: null)","threadName":"Thread-20","applicationARN":<applicationARN>,"applicationVersionId":"1","messageSchemaVersion":"1","messageType":"ERROR"} @timestamp <timestamp> applicationARN <applicationARN> applicationVersionId 1 locationInformation com.amazonaws.glue.catalog.metastore.GlueMetastoreClientDelegate.getFunction(GlueMetastoreClientDelegate.java:1915) logger com.amazonaws.glue.catalog.metastore.GlueMetastoreClientDelegate message software.amazon.kinesisanalytics.shaded.com.amazonaws.services.glue.model.EntityNotFoundException: Cannot find function. (Service: AWSGlue; Status Code: 400; Error Code: EntityNotFoundException; Request ID:<Request ID>; Proxy: null) messageSchemaVersion 1 messageType ERROR threadName Thread-20 ``

Hi I reading from Kinesis and writing into dynamodb. While looping through the records, after 2 hours I get : "Exception has occurred. Broken pipe". After doing some research I added the following code : Aws::SDKOptions options; options.httpOptions.installSigPipeHandler = true; Aws::InitAPI(options); But this was not successful . I did turn on AWS debugger but no specific error came out of the trace. Here is the line that is bringing the error : while (shardIterator != "") { getRecordsRequest.SetShardIterator(shardIterator); Aws::Kinesis::Model::GetRecordsOutcome getRecordsResult = kinesisClient.GetRecords(getRecordsRequest); // <= this line ..... Could you please help ? Thank!

During kinesis video steam using Docker setup when I try to run the Docker (below command) and I am getting the expectation. I want to stream video from PC webcam. How to give Docker access to PC webcam? sudo docker run -it --network="host" --device=/dev/video0 546150905175.dkr.ecr.us-west-2.amazonaws.com/kinesis-video-producer-sdk-cpp-amazon-linux /bin/bash Error:- **docker: Error response from daemon: error gathering device information while adding custom device "/dev/video0": no such file or directory. ERRO0000 error waiting for container: context canceled**

I have a queue in which producers are adding data and consumers wants to read and process it. In the diagram below producers are adding data in a queue with (Px, Tx, X) example (P3, T3,10) here, P3 is the producer ID, T3 is the number of packets required to process and 10 is data. for (P3, T3,10) consumer needs to read 3 packets from the P3 producer so In the Image below, one of the consumer needs to pick (P3, T3,10), (P3, T3,15) and (P3, T3,5) and perform a function on data that just add all the number that is 10+15+5 = 30 and save 30 to DB. Similarly there is a case for P1 producer (P1,T2,1) and (P1,T2,10) sum = 10+1 = 11 to DB. I have read about AWS Kinesis but it has issues, all consumers read the same data which doesn't fit my case. The major issue is how we can limit consumers for: 1 - Read data queue in synchronous. 2 - If one of the consumers has read (P1, T2,1) then only this consumer can read the next packet from the P1 producer (This point is the major issue for me as the consumer need to add those two number) 3 - This can also cause deadlock as some of the consumers will be forced to read data from a particular producer only because they have already read one packet from the same producer, now they have to wait for the next packet to perform add. I have also read about SQS and MQ but the above challenges still exist for them too.  [https://i.stack.imgur.com/7b3Mm.png](https://i.stack.imgur.com/7b3Mm.png) My current approach: for N produces I have started N EC2 instances, producers send data to EC2 through WebSocket (Websocket is not a requirement) and I can process it there easily. As you can see having N EC2 to process N producers will cause budget issues, how can I improve on this solution.

I am using Amazon Kinesis and the [Go SDK for AWS](https://github.com/aws/aws-sdk-go), but I'm getting an error. This is my code: ```go package main import ( "math/rand" "strings" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/session" _kinesis "github.com/aws/aws-sdk-go/service/kinesis" ) func main() { session, err := session.NewSession(&aws.Config{ Region: aws.String("us-east-1"), }) handleErr(err) kinesis := _kinesis.New(session) laugh := strings.Builder{} laughingSounds := []string{"haha", "hoho", "hehe", "hehehe", "*snicker*"} for i := 0; i < 10; i++ { laugh.WriteString(laughingSounds[rand.Intn(len(laughingSounds))]) } _, err = kinesis.PutRecord(&_kinesis.PutRecordInput{ Data: []byte(laugh.String()), PartitionKey: aws.String("laughs"), StreamName: aws.String("laughs"), }) handleErr(err) } func handleErr(err error) { if err != nil { panic(err) } } ``` However, when I run this I get an error: ``` panic: UnrecognizedClientException: The security token included in the request is invalid. status code: 400, request id: dc139793-cd38-fb30-86a3-f92b6410e1c7 goroutine 1 [running]: main.handleErr(...) C:/Users/####/----/main.go:5 main.main() C:/Users/####/----/main.go:34 +0x3ac exit status 2 ``` I have run `aws configure`: ``` $ aws configure AWS Access Key ID [None]: #### AWS Secret Access Key [None]: #### Default region name [None]: us-east-1 Default output format [None]: ``` and the `C:/users/####/.aws/credentials` file is created with the correct configuration. But my program still wouldn't execute successfully. When that didn't work, I also set an environment variable like this: ``` $ $env:aws_access_key_id="####" ``` It still doesn't work. > Version info: ``` $ pwsh -v PowerShell 7.2.2 $ aws -v aws-cli/2.4.27 Python/3.8.8 Windows/10 exe/AMD64 prompt/off ``` OS: Windows 11 (version 21H2). Thanks in advance!

What I'm trying to do: 1. I am using a kinesis data stream to ingest data from a python client sending in JSON's. 2. I setup a kinesis firehose delivery stream with source as the kinesis-data-stream from the previous step and destination as an index on opensearch. I also use a lambda to transform the stream before delivering to opensearch. Now i would like to set the document id for the record I'm transforming in the lambda . I tried setting the key on the transformed record object to be `id` but that creates a document attribute named `id` and sets a value to it. What i would like is to set the _id on the search doc.When i try to set the _id attribute directly by returning it within the transformed record back to firehose delivery stream it generated a destination error : ``` "message": "{\"type\":\"mapper_parsing_exception\",\"reason\":\"failed to parse field [_id] of type [_id] in document with id \\u002749627661727302149016659535428367209810930350656512327682.0\\u0027. Preview of field\\u0027s value: \\u00279b428743-b19a-4fb1-90f2-f132a423b2e8\\u0027\",\"caused_by\":{\"type\":\"mapper_parsing_exception\",\"reason\":\"Field [_id] is a metadata field and cannot be added inside a document. Use the index API request parameters.\"}}", "errorCode": "400", ``` Is there anyway to the doc _id for the documents loaded from firehose to opensearch or would i need to take a different approach by selecting the destination to be a http endpoint and using the rest API's provided by Opensearch (which would kinda be a hassle compared to directly being able to just set the _id attrib) ? What I'm really trying to do is update the indexed documents on a change event. I understand that firehose uses the bulk api from Opensearch , but am unsure about how the upsert operation is handled internally by the kinesis destination connector to opensearch. Hence specifying a fixed id from another DB would be ideal for both insert and update ops to Opensearch in my case.It would be super useful to atleast be able to dictate the type of operation based on some attribute of the kinesis record to opensearch with a reference id for the doc to update.

I am running the specific example "KinesisVideoRekognitionIntegrationExampleTest" from https://github.com/aws/amazon-kinesis-video-streams-parser-library to produce a local "output" to display the bounding box results from amazon rekognition with reference to this set of steps: https://aws.amazon.com/blogs/machine-learning/easily-perform-facial-analysis-on-live-feeds-by-creating-a-serverless-video-analytics-environment-with-amazon-rekognition-video-and-amazon-kinesis-video-streams/ My video is fed properly to KVS ( able to be seen playing on web console) However i am getting this error when running the junit test with nothing displayed on jframe `Running rekognition-kds-stream-application to process stream RekognitionVideoBlog-Stream as worker host.docker.internal:9d3a6f88-50a1-4ab6-8b65-93594b7627ad...14:09:28.173 [pool-2-thread-2] ERROR com.amazonaws.kinesisvideo.parser.examples.GetMediaWorker - Failure in GetMediaWorker for streamName LiveRekognitionVideoAnalysisBlog java.lang.Error: Unresolved compilation problems: 6 The default case is already defined` The log4j is also triggering `java.lang.ClassNotFoundException: com.amazonaws.services.lambda.runtime.log4j.LambdaAppender `

Am testing the docker/rtsp example as provided by: https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/examples-gstreamer-plugin.html how do i edit the "region" such that it targets the KVS i have created in eu-west-1 rather than the default "us-west-2" when running the docker image

I'm stuck developing an application using AWS SDK for Go, and I'm unable to decrypt the Kinesis activity stream records coming from my Aurora MySQL db. Following these steps to decrypt the record payloads, as I try to decrypt the record data key, I get `error:InvalidCiphertextException` I'm currently trying to decrypt the record data key (in order to decrypt the payload) using my KMS key, via this call: `kmsSession.Decrypt(&kms.DecryptInput{CiphertextBlob: []byte(record.Key)})` With `record.key` being the data encryption key from the record I'm reading, as a string, parsed from the whole record. As far as I know, for this usecase there is no need to specify the Encryption Context, or my KMS keyID (although I tried it anyway, getting the same error). I'm opening the kms session as an user with AdministratorAccess policy group rights, and the user is in the Key Users and Key Administrator list for this KMS key. Also, does the record data key needs to be base-64 decoded? Any help is appreciated. Thanks!

Hey, I am currently using boto3's kvam.get_hls_streaming_session_url to get the url of the kinesis video stream as show in this here https://aws.amazon.com/blogs/aws/amazon-kinesis-video-streams-adds-support-for-hls-output-streams/ and am using cv2.videocapture and pass kinesis url to get each frames. But I would like to get the producer and server timestamp of each frame. Is there any way to get it?

Hi, We are currently using Kinesis Firehose to Redshift. I'm looking into migrating to Redshift Serverless. However, it seems there is no support in Firehose to connect to my Redshift Serverless. Will this be added soon? Thanks.

We have a requirements as below this we are doing manually and using MFA token but for production it is not a feasible solution. In the ROSA cluster, we will have a Kubernetes Job that will perform these tasks: 1) download the AWS CLI v2 2) invoke some AWS CLI operations on MSK, S3, KDA (Kinesis Data Application) and maybe OpenSearch. Can you please help how we can achieve this.

I'm working with clients who are locked in to using the AWS SDK for .NET. They are reading records from a Kinesis stream, and I suggested they use [SubscribeToShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShard.html) method because it worked so well in the SDK for Python. But it didn't seem to work in .NET and the above documentation page gives a [404 page](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/SubscribeToShard). Furthermore, the [API documentation for AmazonKinesisClient](https://docs.aws.amazon.com/sdkfornet/v3/apidocs/items/Kinesis/TKinesisClient.html) does not list the method, even though it is referenced in the RegisterStreamConsumer method which seems to be available. Is there any reason why this particular method is not available in the SDK for .NET?

I am currently using [https://github.com/awslabs/amazon-kinesis-video-streams-producer-sdk-cpp]( kinesis-repo) (gstreamer sample) to stream video from rpi to kinesis. The delay is around 10 seconds in the first hour. However, the delay gets built up and goes up to 30 to 40 sec which is not desirable for us. But if I refresh the stream it goes back to 10 sec delay. Could someone help me on how to reduce the delay build up or update the stream if delay is more than 10 sec?

Following the aws blog post : [Dynamic Partitioning](https://aws.amazon.com/blogs/big-data/kinesis-data-firehose-now-supports-dynamic-partitioning-to-amazon-s3/). So I have a firehose delivery stream configured to convert api calls into parquet output to s3 - and now partitioned according to the value of a field from within my data - 'playtype'. However, in some rows the value in this partioning field is null/empty and this causes those rows to error. Therefore I want to adjust the jq expression for the dynamic partioning key to substitute a default value for those null cases. Having trouble with the syntax though. JQ expression specified for the field is .playtype So I have tried things like: ``` .playtype // "some-hardcoded-default" .playtype | if . == null or . == "" then "some-hardcoded-default" else . end ``` These evalute as valid jq in [jqplay](https://jqplay.org/) - but in practice i.e. in firehose are returning a 'jq syntax error'. How can i specify this logic correctly? i.e. partition on this field but if it the value is null or empty then use 'some-hardcoded-default'. Thankyou!

Hey am working on aws kinesis videostreamer and am using kvs_gstreamer_sample. The sample application works good for a day or so and then the stream just hangs and I get the below mentioned error. Could someone help me how to solve this error? > CurlCompleteSync(): HTTP Error 403 : Response: {"message":"Signature expired: 20220128T062720Z is now earlier than 20220128T063533Z (20220128T064033Z - 5 min.)"} >

Hello! Kinesis video stream sdk cpp works well. However, due to internet distruptions sometimes the stream crashes and just hangs(Timeout). > curlCompleteSync(): HTTP Error 0 : Response: (null) > curlCompleteSync(): curl perform failed for url https://kinesisvideo.us-east-1.amazonaws.com/describeStream with result Could not resolve host name :Could not resolve host:kinesisvideo.us-east-1.amazonaws.com Then I get a error code 0x0000000f The operation timed out. Could someone help me how to prevent the stream from timing out when there is internet distruption and reconnect when internet comes back

I used Kinesis Video Streams using Cloud9. When I checked the payment details, there was a Data Transfer item. And there was data transmission to Tokyo and Ireland. Tokyo, I never set up delivery to Ireland, so why did this happen? Does the data transfer to any country happen automatically? below is actual my billing details : AWS Data Transfer APN2-APN1-AWS-Out-Bytes$0.00 $0.080 per GB - Asia Pacific (Seoul) data transfer to Asia Pacific (Tokyo)0.000013 GB$0.00 AWS Data Transfer APN2-EU-AWS-Out-Bytes$0.00 $0.080 per GB - Asia Pacific (Seoul) data transfer to EU (Ireland)0.000004 GB$0.00 Is ec2 charge charged because of cloud9 in this case? Or is it because of kvs?

How am I supposed to process the actual record in Java using KCL? I'm following the guidance provided https://docs.aws.amazon.com/streams/latest/dev/kcl2-standard-consumer-java-example.html, I can connect to the data stream, I can get the number of records available, however what the example doesn't show is how to actually get the record (Json string). From the example I can see that I can use `r.data()` to get the record's data, it comes as a read only `ByteBuffer`, I can convert this to string by using `StandardCharsets.US_ASCII.decode(r.data()).toString()`, however the resulting string is definitely encoded, I have tried doing Base64 decoding but I get error `java.lang.IllegalArgumentException: Illegal base64 character 3f`. So what is the simplest way to get the payload? Below is my `processRecords` method: ``` public void processRecords(ProcessRecordsInput processRecordsInput) { try { System.out.println("Processing " + processRecordsInput.records().size() + " record(s)"); processRecordsInput.records().forEach((r) -> { try { Decoder dec = Base64.getDecoder(); String myString = StandardCharsets.US_ASCII.decode(r.data()).toString(); byte[] bt = dec.decode(myString); } catch (Exception e) { e.printStackTrace(); } }); } catch (Throwable t) { System.out.println("Caught throwable while processing records. Aborting."); Runtime.getRuntime().halt(1); } finally { } } ``` From here I can get `myString` but when I get to `bt` I get the exception shown. I have not found a single resource explaining how to get the record. I post the record to kinesis using `aws kinesis put-record --stream-name testStream --partition-key 1 --data {"somedata":"This Data"}`

I have Fargate Service with FireLens and Fluent Bit. I want to send logs to OpenSearch (or ElasticSearch). Two good options: 1. Firelens / FluentBit -> [kinesis plugin](https://github.com/aws/amazon-kinesis-streams-for-fluent-bit) with **compression and aggregation** -> Kinesis Data Stream -> Kinesis Firehose with Lambda that decompresses and parses logs -> OpenSearch (or ElasticSearch). 2. Firelens / FluentBit -> [es plugin](https://docs.fluentbit.io/manual/pipeline/outputs/elasticsearch) -> Open Search (or ElasticSearch). First version saves bandwidth over the network and custom log parsing can be done. But if no custom log parsing is needed, how can I measure that the second solution is better? Is it better by the way?

Issue: Kinesis Dynamic Partitioning error "errorCode":"DynamicPartitioning.MetadataExtractionFailed","errorMessage":"Non JSON record provided" I am having issues getting Kinesis Dynamic Partitioning to process logs coming from CloudWatch logs after they are transformed via a Lambda function. Current flow: CloudWatch log groups to Kinesis Firehose Delivery stream (with data transformation via Lambda + dynamic partitioning configured) to S3. The logs in S3 are showing "errorCode":"DynamicPartitioning.MetadataExtractionFailed","errorMessage":"Non JSON record provided", however if I turn off Dynamic Partitioning then the logs in S3 are showing correctly as JSON-formatted as per my Lambda function (i.e. {"Account": "123456789012","LogGroup":"<loggroupname>","Log":"<logmessage>"}). The error codes also include the raw data albeit in compressed/encoded form (i.e. "H4sIAAAAAAAAAJVUXXPaO....") however manually decompressing/decoding the data shows data in format below (taken from https://docs.aws.amazon.com/lambda/latest/dg/services-cloudwatchlogs.html) and not the data after Lambda transformation : `{ "messageType": "DATA_MESSAGE", "owner": "123456789012", "logGroup": "/aws/lambda/echo-nodejs", "logStream": "2019/03/13/[$LATEST]94fa867e5374431291a7fc14e2f56ae7", "subscriptionFilters": [ "LambdaStream_cloudwatchlogs-node" ], "logEvents": [ { "id": "34622316099697884706540976068822859012661220141643892546", "timestamp": 1552518348220, "message": "REPORT RequestId: 6234bffe-149a-b642-81ff-2e8e376d8aff\tDuration: 46.84 ms\tBilled Duration: 47 ms \tMemory Size: 192 MB\tMax Memory Used: 72 MB\t\n" } ] }` My understanding of Kinesis Dynamic partitioning is it will process the data after it has been transformed by my Lambda function but it seems this is not the case and it is processing the raw data from CloudWatch logs. Can anybody shed any light on this? Here is the Kinesis terraform code I am using for reference: extended_s3_configuration { role_arn = aws_iam_role.<redacted>.arn bucket_arn = "arn:aws:s3:::<redacted>" prefix = "!{partitionKeyFromQuery:Account}/!{partitionKeyFromQuery:LogGroup}/" error_output_prefix = "processing-errors/" buffer_size = 64 buffer_interval = 300 dynamic_partitioning_configuration { enabled = "true" } processing_configuration { enabled = "true" processors { type = "Lambda" parameters { parameter_name = "LambdaArn" parameter_value = "${aws_lambda_function.decode_cloudwatch_logs.arn}:$LATEST" } } processors { type = "MetadataExtraction" parameters { parameter_name = "MetadataExtractionQuery" parameter_value = "{Account:.Account, LogGroup:.LogGroup}" } parameters { parameter_name = "JsonParsingEngine" parameter_value = "JQ-1.6" } } processors { type = "RecordDeAggregation" parameters { parameter_name = "SubRecordType" parameter_value = "JSON" } } } } } Thanks in advance José

Any idea when aws-kinesis-agent will be updated to use log4j 2.15.0 ?

Hello, I got 404 on POST https://s-######a.kinesisvideo.eu-central-1.amazonaws.com/putMedia request: 1. OPTIONS request: https://user-images.githubusercontent.com/86299738/145280743-18ab38a7-0981-44e1-b3ca-fab52bcca98b.png 2. Main request(which induces the OPTIONS request): https://user-images.githubusercontent.com/86299738/145280462-9bbca7b5-a68d-41fd-aa27-6e2827250a6a.png This documentation https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/API_dataplane_PutMedia.html doesn't state whether the endpoint should handle the OPTIONS request or not. Should I expect a response on the KVS <DataEndpoint>/putMedia OPTIONS request? Detailed info about my attempts in this issue https://github.com/aws-samples/amazon-kinesis-video-streams-demos/issues/143#issuecomment-988313055 Is it still possible to stream to KVS from a browser? Thank you

I know that CDK may have limitation due to CFN having some lag on new features. I am trying to enable Kinesis DS from DDB stream like this console menu. Is this possible from CDK? I searched and eventually gave up. I was wondering if you know this enablement from CDK.

**Traceback (most recent call last)**: File "PATHTOFILE", line 10, in <module> shard_it = kinesis.get_shard_iterator("newstream", shard_id, 'AT_TIMESTAMP', timestamp)\["ShardIterator"] File "C:\Python39\lib\site-packages\boto\kinesis\layer1.py", line 425, in get_shard_iterator return self.make_request(action='GetShardIterator', File "C:\Python39\lib\site-packages\boto\kinesis\layer1.py", line 877, in make_request raise exception_class(response.status, response.reason, boto.exception.JSONResponseError: JSONResponseError: 400 Bad Request {'__type': 'SerializationException', 'Message': 'class com.amazon.coral.value.json.numbers.TruncatingBigNumber can not be converted to an String'} I am making a python app that will collect data from kinesis and write it to a file. I want to read data after a certain period of time. so I am using the "AT_TIMESTAMP" in get_shard_iterator() method. This function will be implemented in lambda.

A customer prefers to receive IoT ingested data to Kinesis Data Stream (KDS afterward). They prefer KDS over Kinesis Firehose (KF afterward) since KDS can support Lambda while KF only supports S3 and non-computing consumers. They want to ingest the KDS so that they can show close-to-realtime analytical dashboards. The little inconvenience with Kinesis Data Stream is that we should predict the number of shards. The IoT application is home appliance, so there will be dynamic usage fluctuation. I searched a bit and found out there is a solution to dynamically adjust the number of shards like this. https://medium.com/slalom-data-analytics/amazon-kinesis-data-streams-auto-scaling-the-number-of-shards-105dc967bed5 But, should I apply this technique? I am wondering if there are already heavy lifting done by others out there. The peak data ingestion is about 60MB / sec. This number is calculated as following. We know that home appliances will not be actively used on night time. There will be about 60K - 100K new devices sold in future. (Since there are already 60K devices sold.) Each device produces about 2 seconds 2KB data. It makes the peak about 60K * 1KB. If we do the batch done correctly, I assume that it should consume about 60 shards.

Hi, I am trying to understand how KCL manages total number of RecordProcessors when there are multiple instances of workers(or consumers). So I did have a look at below link, and the example in the doc makes sense. https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-scaling.html But have a few follow up questions just to make sure if my understanding is correct. In the example, it says # of "record processors" will automatically be updated per KCL worker. And my question is: **So if I increased the same KCL worker (same application name and same kinesis configs), KCL will figure out how many workers (or instances) are out there based on the application name, and readjust # of record processors for each KCL worker accordingly.** Thanks in advance!

Need to compare preferred choice between Kinesis Firehose and IoT Core for streaming data ingestion. From my observation, * IoT Core is more suitable to infrequent, bi-directional and network limited IoT data ingestion from many IoT devices. * Kinesis Firehose is more suitable for ingesting constant and quite big stream of data to AWS cloud. Then my question is "how big"? For example, if a customer's IoT end device is making 1MB/s or 5MB/s (or even arbitrary X MB/s) of read-only constant sensor data stream which needs to be ingested to AWS Cloud. Should they consider IoT Core or Kinesis Firehose/Data-Stream? Let's assume that the data to be sent will be quite well JSON formatted and AWS Cloud will save that data to S3 directly. What's the threshold X MB/s value to determine whether to use IoT Core or Kinesis? Thanks!

Hi, I have a customer who wants to write solar power generator's sensor data to S3. The data stream happens usually during the day time and almost no data on night time. It will likely be about 1MB / second during the day time. It may vary to 5MB or more depending on how many solar panels in the deployed generator area. There may be network off time to time since solar power generators are usually place on mountain area. They want to save the sensor data to S3 since it's all read only data there. They will use SageMaker as well for complex Machine Learning process. The ML process + weather information will eventually make prediction about how much power will be generated for the next month after power generation commitment is made. There is no control data going back to edge side, so I filtered out IoT Core from the data ingestion consideration. There were similar previous project in Korea using IoT Core, but had trouble from streaming data to the cloud and found rather Kinesis was better approach. However, in the later stage, when there will be control data going back to the edge side, Greengrass or IoT Core will be considered for non-stream data. The customer and I would like to know which of the following (or some new method) would be the best approach. - Directly writing to S3 using CLI (or other method) would be worthwhile since S3 writing directly is free. I never observed any projects or architecture diagram writing to S3 directly. So I answered to the customer that this is unlikely, but they demand why which I do not know at this moment. - Writing to S3 using Kinesis Data Stream and turning the stream shard off on night time. Currently, this is my best bet, but I would like to know your opinion. - Using AWS File Gateway to write to S3. But I think this is not worthwhile since local gateway does not need to access the cached files. It's just one way to S3 from the sensors. Could you please share your opinion? Thank you!

I have Kinesis consumer code that does a DescribeStream and then spins up a new Java thread per shardId to consume off each shard. I get 8 shardIds when I've only configured 3 in my Stream. Why is that? I don't want to have 5 extra threads consuming constantly and getting zero records. Below, you can see I'm logging the total # of records processed on each shard. 2020-11-19 08:59:49 INFO GetRecords:109 - # Kinesis consumers: 8 2020-11-19 08:59:49 INFO GetRecords:112 - Kinesis - ShardId: 'shardId-000000000000', Total Records: 0 2020-11-19 08:59:49 INFO GetRecords:112 - Kinesis - ShardId: 'shardId-000000000001', Total Records: 0 2020-11-19 08:59:49 INFO GetRecords:112 - Kinesis - ShardId: 'shardId-000000000002', Total Records: 0 2020-11-19 08:59:49 INFO GetRecords:112 - Kinesis - ShardId: 'shardId-000000000003', Total Records: 19110 2020-11-19 08:59:49 INFO GetRecords:112 - Kinesis - ShardId: 'shardId-000000000004', Total Records: 0 2020-11-19 08:59:49 INFO GetRecords:112 - Kinesis - ShardId: 'shardId-000000000005', Total Records: 0 2020-11-19 08:59:49 INFO GetRecords:112 - Kinesis - ShardId: 'shardId-000000000006', Total Records: 18981 2020-11-19 08:59:49 INFO GetRecords:112 - Kinesis - ShardId: 'shardId-000000000007', Total Records: 16195 **Background:** I started with 1, then configured 2, then, 3. Does this have something to do with the other shardIds that have 0 records? If so, what is the recommended code/practice to ignore a certain type of shard?

I'm following tutorial from https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/hls-playback.html#how-hls-ex1 It works great! But the access key is visible to user in the source code. How can I go about hiding access key from the user? Tutorial puts it visible in the source code: var options = { accessKeyId: $('#accessKeyId').val(), secretAccessKey: $('#secretAccessKey').val(), sessionToken: $('#sessionToken').val() || undefined, region: $('#region').val(), endpoint: $('#endpoint').val() || undefined } var kinesisVideo = new AWS.KinesisVideo(options); var kinesisVideoArchivedContent = new AWS.KinesisVideoArchivedMedia(options); I'm using Django web server.

I have a Kinesis stream which feeds into a Firehouse stream which outputs ORC to S3. All this has been working fine for about 2 years until about 2 days ago, when a new Athena table appeared named "format_conversion_failed (Partitioned)" And my primary Athena table that I use for my production data STOPPED being populated! In the "format_conversion_failed (Partitioned)" table I see a bunch of rows with all the same messages: DataFormatConversion.SchemaValidationTimeout Timed out while retrieving table from Glue. If you have a large number of Glue table versions, please add 'glue:GetTableVersion' permission (recommended) or delete unused table versions. If you do not have a large number of tables in Glue, please contact AWS Support. I do not have a large number of Glue Tables (14). (2 are newly created by this error problem). Can somebody point me to how I can resolve this problem?

I have set up a firehose (direct put) to deliver messages to a s3 bucket. Unfortunately, regardless of the number of message to be delivery, Firehose always delivers n-500 messages. In screenshot attached, we can see what happens when attempting to deliver 500000 records, it gets stuck at 4995000. Same happens when I attempt to deliver 200000 messages, it gets stuck at 199500 I have replaced the firehose, change the instances, even tried to do the same from different aws accounts, but the result is the same. Anyone who could shed some light on this would be very much appreciated. Sincerley, Josep Maria Edited by: JosepMaria on Sep 27, 2020 5:53 AM Edited by: JosepMaria on Sep 27, 2020 5:57 AM

Hi all, from what I understand there is currently no option to use data analytics to process stream via an enhanced fan-out. Any plans on introducing it? Thanks, Aleksander Edited by: aleksandersss on Sep 21, 2020 3:45 PM

A customer wants to use Kinesis for gathering and aggregating log data from multiple accounts into a central account. They have two destinations - 1. S3 (using parquet transformation for easy Glue/Athena access), 2. ElasticSearch (for visualization and common queries). The question is what is the AWS recommended approach for this and why? These are the three approaches that come to mind, however looking for more guidance: A. Using Kinesis Data Stream as the primary delivery mechanism, and use that to feed two Kinesis Firehose streams - each one targeted at one of the destinations (see above). B. Using Kinesis Firehose as the primary delivery mechanism, targeting initial delivery into S3 (with parquet conversion), and using an S3/Lambda trigger to load the data into ElasticSearch. C. Using Kinesis Firehose as the primary delivery mechanism, targeting both S3 (raw - not converted) delivery and ElasticSearch delivery. Then using an S3/Lambda trigger to transform the raw S3 data into parquet format saved back into S3.

A customer is using dynamoDB streams to pass events to a Kafka topic. They would like to use lambda to poll the stream, but the consumer (kafka cluster self managed in Kubernetis) is in a different account. I see that the lambda that reads from DDB needs to be in the DDB account. As a workaround customer is considering this- https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.KCLAdapter.html (deployed in the Kafka account in the kubernetis cluster) but would prefer have a serverless solution instead. I found this- https://aws.amazon.com/premiumsupport/knowledge-center/lambda-cross-account-kinesis-stream/ Will this mean the traffic goes through public internet (not desired) Is there a work around using VPC peering / VPC sharing / Private link Note - The Kafka cluster can be in the same region, but sometime is in a different region so solution should support both.

A data scientist is looking to host a Tensorflow model in SageMaker and process low volume streaming event data (~2-3 per second) to collect inferences about each event. Data scientist is looking at having the SageMaker inference model plugged in as a Kinesis Data Analytics Application but Kinesis Data Analytics currently only supports SQL or Flink. One option to set up an ECS or Lambda service to consume data from Kinesis or SNS and invoke the SageMaker inference endpoint per message, but if there is a more automated and optimal solution available for these kind of workflows. It is not possible to pass multiple requests currently to a SageMaker endpoint, yet Tensorflow models tend to perform much better on batches of data rather than multiple single invocations so some windowing would be beneficial. Ideally the client would want to react to an inference within 10-15 seconds of the event being processed so an S3 based batch approach is probably too slow. Is there anything you can recommend for handling this sort of workload?

Hello, I'm using Amazon CONNECT service and I need to record the whole audio, including when the caller is talking with a LEX bot. Therefore I've enabled streaming, inserted the start stream block, and I'm properly get 1 stream in Kinesis Video Streams per call. BUT from here I cannot figure out HOW to get the record back to me in a file neither at least being able to PLAY them remotely: - the online AWS media playback falls in an "Unexpected error - The type of the media is not supported or could not be determined from the media codec ids: (track 1: A_AAC), (track 2: A_AAC)" - through API / CLI with GetClip, I'm properly using the syntax but it falls in the same error: << An error occurred (UnsupportedStreamMediaTypeException) when calling the GetClip operation: The type of the media is not supported or could not be determined from the media codec ids: (track 1: A_AAC), (track 2: A_AAC).>> My AWS CLI: aws kinesis-video-archived-media get-clip --stream-name <name of the stream> --clip-fragment-selector FragmentSelectorType="SERVER_TIMESTAMP",TimestampRange={StartTimestamp=2020-09-02T23:44:12_0200,EndTimestamp=2020-09-02T23:45:00_0200} --endpoint-url <the endpoint retrieved from get-data-endpoint for GetClip API> <filename> I've also tried the GetMedia API (with CLI aws kinesis-video-media get-media + proper parameters), I retrieve a file containing the list of the chunk... but I don't know how to read the audio from it (even VLC failed, and it's normal since it contains data attributes...). To me actually, the Amazon Connect stream recording service is totally unusable. Thanks for your help and support, every help and advice are welcome. Stephane

A customer is looking at increasing consumers for our Kinesis Stream. Currently they have a consumer script that polls every min, which they want to add firehose to consume as well. What are the constraints and limits that they should be looking out for, and how to calculate the optimal number of shards required to support the increased number of consumers? Is this approach even correct?

A customer is currently crawling the web and writing to S3 + loading to Amazon Elasticsearch. The current architecture looks like this. ECS -> EC2 -> S3 -> EC2 -> ES Looks like they set this up a long time ago, and would like to optimize the workflow. They are using ECS (they have containerized old code written in C#), running that through EC2 to write to S3, and then using another EC2 instance to load data into ES. This seems like a very inefficient solution. There also seems to be a lot of unknown around ECS and crawler code, and they would not like to touch it. Given this information, I thought the following architecture would make more sense. ECS + Kinesis agent -> Kinesis Firehose + Lambda -> ES (and S3 through backup) What I would like to know is 1) Does the above workflow make sense? 2) Customer is concerned about price. While it is difficult to get an exact estimate, I was hoping to get some insight into if anyone has done a similar comparison. This customer has a small team and would not like to commit resources to end up with a more expensive solution.

Hi there, I am currently working with Kinesis shards and I would like to make sure that the hashing function is identical to another message bus I am using. What this entails is I need to write a custom hash function for that other message bus. I have read the following documentation for Kinesis: <https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html> <https://docs.aws.amazon.com/cli/latest/reference/kinesis/describe-stream.html> My understanding from reading the documents is that each shard out of n shards starting from 0 will have a range of keys 2^128/n and each event's partition key will be hashed out to a 128-bit integer via md5 hashing. Eg. **For n partitions (starting from 0):** Partition0 key range: Starting Hash: 0 Ending Hash: (2^128 / n) - 1 Partition1 key range: Starting Hash: (2^128 / n) Ending Hash: 2*(2^128 / n) - 1 ... ... Partition(n-2) key range: Starting Hash: (n-2)*(2^128 / n) Ending Hash: (n-1)*(2^128 / n) - 1 Partition(n-1) key range: Starting Hash: (n-1)*(2^128 / n) Ending Hash: 2^128 - 1 Is there any official documentation or references around this method? I want to make sure my understanding above is correct so I can ensure that the custom method I implement is identical to the Kinesis partitioning mechanism. Edited by: VictorL85 on Apr 23, 2020 5:12 PM

I&#39;m currently looking at how to build our WAFv2 configuration using CloudFormation. Creating the WebACL and adding rules is fine, but I can&#39;t see any option to apply the logging configuration (Logging and Metrics > Logging in the console). If my Kinesis Data Firehose already exists, then I want to configure each new WebACL to log to this stream, but I can&#39;t see any option to configure this? If we can configure it via the console then I would presume that it can be done via CloudFormation. Has anyone been able to do this? I must be missing something obvious! Thanks, Chris

Hi: All, I am using a one shard kinesis data stream, and bit confused/curious about the behavior of AWS KCL. I use KPL version 0.13.1 to write to the stream and KCL version 1.11.2 to read from the stream. According to Kinesis doc https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-implementation-app-java.html "**All workers associated with this application name are assumed to be working together on the same stream.**" I got it, but what I have noticed that when you fire 2 instances of consumer app using KCL with same name working on the same stream with one shard, only first instance received the data from the stream, and the other instance sits idle. The second instance only receives data, if I kill or stop the first instance. Is this the expected behavior? If yes is there any way of telling the KCL to load balance the data to different instances automatically? If it is not possible for one shard, is it even possible for multiple shards?

I want to be able to import all emails being received on an external SMTP server for an externally managed domain(say waffles.com). 1. How do I ingest all the emails being received to all users of that waffles.com in my AWS S3 bucket? Assume that I am able to create a new user in that SMTP server with Admin right - how do I create a connection to that SMTP server so that all emails can be dumped in my bucket for post processing? Really appreciate if anyone can share any insights on what services to use, or any other idea that can help me achieve this. Thanks.

Hi, My customer is looking for an architecture to use the Aurora Database Activity Streams feature to provide some Database Activity Monitoring (DAM) capability. They investigated the Imperva 3rd party offering, and concluded it didn't scale and they wanted something serverless using Kinesis and Lambda. Has anyone got experience with such a setup? My customer wants to know how to best build this and how much it would cost for a given number of invocations. Their current envisaged setup is: DAS -> Kinesis Stream -> Kinesis Firehose -> S3 -> Lambda -> SNS (for alerts into Splunk). Thanks

Events are sent once from a lambda function to a firehose delivery stream without any transformation or conversion. observing the data on s3 **every** record is written twice. the lambda uses "putRecordBatch" to send the object to FDS. example: ``` {"e":"pv","url":"https://foo.com", "id":"1552c074-8432-43e9-a1e5-7e5206ed8790"} {"e":"pv","url":"https://baz.com", "id":"ebb18ce2-fe8d-4c1b-8d2e-3338762c2c8a"} {"e":"pv","url":"https://foo.com", "id":"1552c074-8432-43e9-a1e5-7e5206ed8790"} {"e":"pv","url":"https://baz.com", "id":"ebb18ce2-fe8d-4c1b-8d2e-3338762c2c8a"} ``` "id" is unique I understand that duplication might happen but not to 100% of the data. how can we figure out the cause of this duplication if it is abnormal.

Hi everyone! Could someone explain me this related to kinesis DS limits found in the documentation: [https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html][1] 1. GetRecords can retrieve up to 10 MiB of data per call from a single shard, and up to 10,000 records per call. Each call to GetRecords is counted as one read transaction. 2. Each shard can support up to five read transactions per second. Each read transaction can provide up to 10,000 records with an upper limit of 10 MiB per transaction. 3. Each shard can support up to a maximum total data read rate of 2 MiB per second via GetRecords. If a call to GetRecords returns 10 MiB, subsequent calls made within the next 5 seconds throw an exception. I find these points not very consistent and maybe contradictory. Did I miss something? I appreciate your help. [1]: https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html

I see on the Kinesis Data Analytics for Java Applications Limits page that "You can create up to 50 Kinesis Data Analytics applications per Region in your account. You can create a case to request additional applications via the service limit increase form." I take it that this means that 50 applications is a soft limit. If I need to request an increase, is there a hard limit on the number of applications that I can request? Thanks.

Is it possible to put records into a kinesis stream located in another AWS account? I configured my kinesis producer with an STS assumed role, but I can't see how to specify a stream name in another account (it doesn't accept an ARN, only a stream name). Init logic: ``` AWSCredentialsProviderChain credentialsProvider = new AWSCredentialsProviderChain( DefaultAWSCredentialsProviderChain.getInstance(), new STSAssumeRoleSessionCredentialsProvider.Builder( config.getKinesisStreamRoleArn(), roleSessionName) .withExternalId(config.getKinesisStreamRoleExternalId()) .build()); Properties props = new Properties(); props.load(getClass().getClassLoader().getResourceAsStream("kinesis_producer_config.properties")); KinesisProducerConfiguration producerConfig = KinesisProducerConfiguration.fromProperties(props) .setCredentialsProvider(credentialsProvider); if (!Strings.isNullOrEmpty(config.getKinesisStreamRegion())) { producerConfig.setRegion(config.getKinesisStreamRegion()); } return new KinesisProducer(producerConfig); ``` and the error: ``` Encountered AWSError ResourceNotFoundException Stream my-stream under account 111111111111 not found.: ``` (where 111111111111 is the other aws account)

A customer currently use Kinesis (I guess data stream) in their own IoT service. Customer do believe that KCL recorded how handling data when EC2 is being scaled in or terminated. (There is a possibility of terminating EC2 when the data has been handled) **Q1)** They wonder how does this record handle? Or **Q2)** They have to set lifecycle hook when EC2 is being scaled in or there is something that KCL managed this record by itself? Customer would like to have any example or practice.

A customer would like to process streaming data in Spark Streaming using **PySpark and would like to output the results on a Kinesis stream**. While PySpark does support reading data from Kinesis (http://spark.apache.org/docs/latest/streaming-kinesis-integration.html), I do not see any support for writing data to Kinesis. DataBricks has some documentation for creating a Kinesis sink for Spark (scala) (https://docs.databricks.com/spark/latest/structured-streaming/kinesis.html), but if I understand the documentation correctly this is based on the `ForeachSink` which is not supported in PySpark (http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach) > The foreach operation allows arbitrary operations to be computed on the output data. As of Spark 2.1, this is available only for Scala and Java. To use this, you will have to implement the interface ForeachWriter (Scala/Java docs), which has methods that get called whenever there is a sequence of rows generated as output after a trigger. Note the following important points. Has anybody come across a way to achieve this?

A customer is asking how egress data from Kinesis data stream to his on-premises consumer is charged. The [Kinesis pricing page][1] indicates: "Data transfer is free. AWS does not charge for data transfer from your data producers to Amazon Kinesis Data Streams, or from Amazon Kinesis Data Streams to your Amazon Kinesis Applications." Based on this I would assert that all data transfer to and from Kinesis is free. I am somewhat vary though that this may not apply across regions and/or from on-premises. Can someone please clarify for me, while I go back to the customer to get a better understanding of their approach. Having remote producers and applications may inject complete different challenges altogether. [1]: https://aws.amazon.com/kinesis/data-streams/pricing/

A customer in need of configuring an API Gateway fronting a Kinesis Stream. We're currently working through the docs to [Create an API Gateway API as an Amazon Kinesis Proxy][1]. For the `PutRecord` and `PutRecords` APIs, customer wants to support the following payload structure (note the data-elements are passed in as json): { "records": [ { "data": <json>, "PartitionKey": <String> }, { "data": <json>, "PartitionKey": <String> } ] } e.g. like the following: { "records": [ { "data": { "Events": [ {"DSType":"RTDS","DSInstance":"Unit 1","DSPoint":"Tag 1","TimeStamp":"2017-10-10T11:10:00.0000000+02:00","Value":8.0,"Quality":192}, {"DSType":"RTDS","DSInstance":"Unit 2","DSPoint":"Tag 2","TimeStamp":"2017-10-10T11:10:00.0000000+02:00","Value":7.0,"Quality":193} ], "Project":"EventAcquisitionStream", "Plant":"Plant1" }, "partition-key": "some key" }, { "data": { "Events": [ {"DSType":"RTDS","DSInstance":"Unit 3","DSPoint":"Tag 3","TimeStamp":"2017-10-11T11:10:00.0000000+02:00","Value":8.0,"Quality":192}, {"DSType":"RTDS","DSInstance":"Unit 4","DSPoint":"Tag 4","TimeStamp":"2017-10-10T11:10:00.0000000+02:00","Value":7.0,"Quality":193} ], "Project":"EventAcquisitionStream", "Plant":"Plant2" }, "partition-key": "some key" } ] } Per AWS documentation referenced above, we have configured the following model and body template mapping for the `PutRecords` API: Model: { "$schema": "http://json-schema.org/draft-04/schema#", "title": "PutRecords proxy payload data model", "type": "object", "properties": { "records": { "type": "array", "items": { "type": "object", "properties": { "data": { "type": "object", "properties": { "events": { "type": "array", "items": { "type": "object", "properties": { "dstype": { "type": "string" }, "dsinstance": { "type": "string" }, "dspoint": { "type": "string" }, "timeStamp": { "type": "string" }, "value": { "type": "number" }, "quality": { "type": "integer" } } } }, "project": { "type": "string" }, "plant": { "type": "string" } } }, "partition-key": { "type": "string" } } } } } } Body template mapping: { "StreamName": "$input.params('stream-name')", "Records": [ #foreach($elem in $input.path('$.records')) { "Data": "$util.base64Encode($elem.data)", "PartitionKey": "$elem.partition-key" }#if($foreach.hasNext),#end #end ] } Problem is, when this is decoded on the Kinesis consumer-side, the contents of `$elem.data` has been transformed into invalid json, e.g. replacing `:` with `=` (note for `Events`, `Project` and `Plant` elements): Decoded payload: {Events=[{"DSType":"RTDS","DSInstance":"Unit 1","DSPoint":"Tag 1","TimeStamp":"2017-10-10T11:10:00.0000000+02:00","Value":8.0,"Quality":192},{"DSType":"RTDS","DSInstance":"Unit 2","DSPoint":"Tag 2","TimeStamp":"2017-10-10T11:10:00.0000000+02:00","Value":7.0,"Quality":193}], Project=EventAcquisitionStream, Plant=Plant1} What is going on here, how can I configure my body template mapping to retain the json data-structure within the data elements of the request, while still allowing for base64-encoding? [1]: http://docs.aws.amazon.com/apigateway/latest/developerguide/integrating-api-with-aws-services-kinesis.html#api-gateway-get-and-add-records-to-stream

A customer is interested in re-factoring their credit-card processing app, and they want to use "streams" to glue the various parts together. They are considering either Kinesis or Kafka. Besides the fact that Kinesis is not PCI compliance (client side encryption), they are concerned about the 5 read TPS limit per shard. That seems somewhat low for a real-time app. The customer only has 100-200 ms time budget to process the transaction. While I know that Kinesis can pull multiple records per read, this implies a potential 200ms gap between reads. While not a lot, this may be a show-stopper for my customer. Is there any way to make Kinesis more responsive (higher read TPS), or is this gap a non-issue in terms of "real-world" latency?!

A customer is looking to build a message processing system using Kinesis. This relative low volume (so only one shard will be needed), but low-ish latency is important (<1 sec end-to-end). Currently, there would be four consumers of the stream. However, the expectation is that more consumers would be added in the future. Do we have best practice guidelines on fanning-out Kinesis streams to multiple consumers?

Are there any third-party tools that customers can use to operationally manage Kinesis streams? For example, if a customer has dozens of Kinesis streams, outside of using AWS Console and scripting against CloudWatch, does a graphical tool exist to help with the operational burden of managing multiple Kinesis streams.

I’m aware that the items in the stream are ordered. My principle question then is about **whether or not the shards are ordered in the stream description**. If they are, I can always just grab the latest shard and process records from there; if not, I would need to keep track of the shards already processed in order to get only the new shard from the next iteration when I grab the stream description—I.e., after the current shard is closed and I have to quit out of the getRecords loop for that shard. Further Context: I have already complete a KCL implementation, but am looking at doing a lower-level implementation, as that looks to be more suited to our particular needs. Ultimately, I just need to get the latest record from the table (broadcast amongst all of the service nodes). That said, my quick question is as follows: **is the shard list returned from DescribeStreamResult.getShards() an ordered list?** (I presume this follows the general describe stream API, but have yet to see anything indicating whether or not this list is ordered.)