Questions tagged with Amazon Kinesis Data Analytics

Hello, I have Change Capture Data landing into a S3 bucket. The CDC data is from an AWS RDS. This is my starting point. Now, let's say, I have tables taba, tabb, tabc. I have the 3 folders for each table and all the CDC data is segregated per table in these folders. Now I must create another bucket, that will only hold the latest data. so in taba folder, if I had rows with 10 inserts , 10 updates and 3 deletes. How do I bring it to a state of finalization in the new bucket, meaning I should only have the latest data?lg...

Getting the below error SYNTAX_ERROR: line 1:15: Schema ecommerce-raw-apsouth1-84245-dev does not exist This query ran against the "" database, unless qualified by the query. Please post the error message on our forum or contact customer support with Query Id: 57acf945-9593-4a96-8c8e-58f1e5be24b3lg...

Currently I am trying out a KDA + Flink setup and first thing that stops me a little bit is failover strategy. There is only one option for a failover: take savepoint -> destroy cluster -> start new cluster -> start job from savepoint, and this option ends up in few minutes downtime. I am aware that aws uses a bit different strategy for maintenance window where both clusters run at the same time and only job stop, start creates a downtime which is way shorter then the only available option. Does anyone know if this and when this second option will be available for cases like autoscale/ user maintenance/ etc.? And will it be compatible with custom autoscaling rules?lg...

I saw that in Flink 1.15, this was added(https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/ops/state/task_failure_recovery/) but I need to handle the scenario of my source and sink(which are Kinesis Data Streams) going away(and returning a UnknownHostException). I know it's not a thing that will happen very often with KDS but we happened to run into this one time. Our Flink App is written in Java. I see several config constants but was thinking about making this exception recoverable and using this consumer config(SHARD_GETRECORDS_RETRIES). I'm not sure on the producer side(not sure if the KPL will do the retry for me).lg...

I have an event processing architecture that's basically KPL Producer -> Kinesis Data Streams(Raw) -> KDA(Flink Aggregation) -> Kinesis Data Streams(Agg) -> Lambda -> Timestream . With a basic setup of 10 Raw Shards, 2 KPU's on the Flink App, and 5 Agg Shards(On Demand) and maybe 100 requests per second, my stack processes the data with maybe a delay of minute going through the whole workflow to Timestream. As I pre-scale things up for a much higher workload(that won't be arriving for several days), I notice that my current delay goes from 1 minute to 30 minutes to arrive in Timestream. The question I need to understand is where in my workflow is the latency being added. My current stack is now at 200 Raw Shards, 25 KPU's, and 15 Agg Shards(On Demand). I know the delay is most likely in Flink as I can see bursts of records being written out every 30 minutes using the incoming record count of my Agg KDS but my question is why? I know that it will clear up when my requests per second go thru the roof but I always wondered what underneath the hood is occurring in Flink to add this latency.lg...

Hello, I'm trying to figure out the best way to output messages to SNS from a Flink Kinesis Analytics application. My understanding is firehose has ~60s of latency, which is not acceptable for this use case, as we want <1 sec of latency from ingestion to output. If I'm ok with > once delivery of output, could I just create a flink function to send a message to SNS directly and not have to worry about creating a sink that handles checkpoints/deduplication/batching/whatever else? We're looking at <1000 TPS. Any thoughts or suggestions would be appreciated. Thanks!lg...

Hello, I am having trouble to properly handle query with tumbling window. My application sends 15 sensor data messages per second to Kinesis Data Stream, which is used as an input stream for Kinesis Analytics application. I am trying to run an aggregation query using a GROUP BY clause to process rows in a tumbling window by 60 second interval. The output stream then sends data to a lambda function. I expect that the messages should arrive at lambda every 60 seconds but instead, they arrive much faster, almost every second, and the aggregations don't work as expected. Here is the CloudFormation template that I am using: ApplicationCode: CREATE OR REPLACE STREAM "SENSORCALC_STREAM" ( "name" VARCHAR(16), "facilityId" INTEGER, "processId" BIGINT, "sensorId" INTEGER NOT NULL, "min_value" REAL, "max_value" REAL, "stddev_value" REAL); CREATE OR REPLACE PUMP "SENSORCALC_STREAM_PUMP" AS INSERT INTO "SENSORCALC_STREAM" SELECT STREAM "name", "facilityId", "processId", "sensorId", MIN("sensorData") AS "min_value", MAX("sensorData") AS "max_value", STDDEV_SAMP("sensorData") AS "stddev_value" FROM "SOURCE_SQL_STREAM_001" GROUP BY "facilityId","processId", "sensorId", "name", STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND); KinesisAnalyticsSensorApplicationOutput: Type: "AWS::KinesisAnalytics::ApplicationOutput" DependsOn: KinesisAnalyticsSensorApplication Properties: ApplicationName: !Ref KinesisAnalyticsSensorApplication Output: Name: "SENSORCALC_STREAM" LambdaOutput: ResourceARN: !GetAtt SensorStatsFunction.Arn RoleARN: !GetAtt KinesisAnalyticsSensorRole.Arn DestinationSchema: RecordFormatType: "JSON" I would really appreciate your help in pointing what I am missing. Thank you, Sergelg...

Hi, we are using X-Ray for application logs tracing but X-Ray is not able to show case all the logs and after some times it drops off the tracing. Is it some limitations with X-Ray or how we can enables X-Ray for the complete logs tracing.lg...

We have several desktop (Win/Mac/Nix) monolithic applications that generate JSON events exposing these events via POST. Currently it is a one way deal, outbound from app to the stream for nothing but analytics. We are now looking to analyze those events to determine if there are actionable "data sequences" that start to appear. **My Question is (and opinions are welcomed...)** **What is the most efficient (#1 - performant, #2 - cost effective, #3 - resilient) way to stream those records (250-500 byte msg, at a peak of 1,000 / second) to AWS for the purpose of:** 1. Transforming the record data to a known model (i.e. std date format, simple XForms) 2. Analyzing the normalized record (ML - predictive analytics or iterative learning) 3. Persisting the record 4. Based on the analysis, sending back a payload (or pointer to a CDN stored payload) to the original calling app? In a nutshell looking to do a type of on-preises, *near real-time* application performance analytics of the applications using a single on-premises service spewing JSON out to AWS native cloud services that do all teh work, and return info as timely as possible. Thanks. <><lg...

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}} ```lg...

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 ``lg...

Hi All, I've started learning AWS Data Analytics. While going through study material I have found use of Java code/programming. Can someone please guide me what level of Java expertise needed to get expertise in AWS data analytics?lg...