Questions tagged with Amazon Kinesis Data Analytics

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

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

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 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?

My application writes text data to a Kinesis Stream using pipe symbol as field separator. I've defined a table in Glue and specified CSV with the | separator. It's defined as a 3 column table and each column is string. However, when running interactive queries in Zeppelin using %flink.ssql, it chucks the following errors: Caused by: java.io.IOException: Failed to deserialize CSV row 'DBSERV-1-REPO1|2022-02-24 11:55:34|WARNING: Heavy swapping detected '. Caused by: java.lang.RuntimeException: Row length mismatch. 3 fields expected but was 1. If I test writing to Kinesis using comma as separator and set the Glue table as CSV with comma separator, the query works correctly and I can read the data. Do I have to separately instruct the Zeppelin Note when the separator is not a comma?

We are trying to run a Query on AWS Kinesis, and use autoscaling. We found that if user specified the Parallelism option in AWS Kinesis configuration, AWS Kinesis will apply the specified parallelism to all the operators. We also tried run the query without specifing any parallelism (set ConfigurationType to DEFAULT). And in this case, Kinesis will only start 1 task manager with 1 slots, even if that's not enough to run our Query. So we are wondering does AWS Kinesis auto scaling supports setting different level of parallelism for different operators? Thanks

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

We've been using DDB streams as an event source for lambda functions serving our analytics pipeline. Additionally, we configured the lambda to send the failed records metadata to a DLQ in case the processing fails after a given number of retries - by doing so, we get the event metadata rather than the actual failing records. Once we fix the underlying issue we currently have no easy way to reprocess the failing records. Looked at [KCL](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-kcl.html) but it requires Java to be installed. As we use Python, we find it somewhat strange to install Java just for the sake of getting the impacted DDB records. To me, the data that's sent to the DLQ should include the impacted records. Would appreciate any easy way to fetch the impacted records given the metadata sent in such a case.

Is it possible to create multiple Kinesis Data Analytics apps and use the same source which is a Kinesis firehose delivery stream? Are there any throttling concerns for the Firehose stream when multiple Analytics app polls data from it?

Can a Lambda data transformation function attached to a Kinesis Firehose perform a join between the streaming data in that Kinesis Firehose and data in S3 through Athena, and pass the results to Kinesis Analytics for further processing? Is this functionality supported?

I am interested to build a real time application with data residing on their Kafka cluster. The Kafka is running on AWS, using K8S not using MSK. What are the step required to integrate Kinesis Data Analytics Flink application with their Kafka cluster?

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?

A customer is building a streaming solution that processes messages from their customers. Data in the stream is sent from multiple customers - each JSON in the stream has a customerId that tells you what customer originally sent the data. They want to store a copy of this data on S3, but partitioned by customerId. We *could* do this with multiple Firehoses (one for each customer). Is this possible to do with Flink - so use Flink to pull from the Kinesis stream and send to multiple destinations on S3, into files based on customer Id? I'm guessing this is possible - is it easy and can we use any existing libraries? Thanks