Interaction between EMR Serverless and Kinesis

Question

I have been experiencing quite a few problems with Spark and Kinesis, and wanted to clarify my understanding of it.  Say I have the following logic:

Subscribe to stream 
    Event arrives and RDD is built
    Transformation 
    Transformation
    Transformation
    ForEachBatch
       Action

It seems that when I do the ACTION, Spark goes back to Kinesis and asks for the actual batch of records using lazy evaluation, potentially from multiple executors.

I would expect the mechanism to be that an RDD is created in memory earlier in the process.   I have tried explictly caching the RDD, but I can only do that inside the streaming context when it is too late.

This breaks my mental model of streaming as it seems Kinesis is telling us an event is ready, then we have to go back and ask for it when the action is processed.

A few specific questions:

- Can this model not result in multiple executors and tasks connecting to Kinesis in parallel to pull batches, meaning that 1 inbound batch of events form Kinesis turns into tens or hundreds of calls back into Kinesis from different executors?

- The connection to Kinesis takes up to 5 seconds to instantiate.  Rather than do it once, this model seems to imply that our executors repeatedly need to connect back to Kinesis to process actions.  This also seems inefficient and slow.

The implication of this is that our Kinesis and Spark setup is way too slow and unscalable due to various interaction with Kinesis which I didn't expect to see.

Any guidance what is happening here or sources of documentation?  Thanks in advance!

Ltlaksh · Answer

Amazon Kinesis Data Streams connector : https://docs.aws.amazon.com/emr/latest/EMR-Serverless-UserGuide/jobs-spark-streaming-connectors.html

The Amazon Kinesis Data Streams connector for Apache Spark enables building streaming applications and pipelines that consume data from and write data to Amazon Kinesis Data Streams. The connector supports enhanced fan-out consumption with a dedicated read throughput rate of up to 2MB/second per shard. By default, Amazon EMR Serverless 7.1.0 and higher includes the connector, so you don't need to build or download any additional packages. For more information about the connector, see the spark-sql-kinesis-connector page on GitHub

The following is an example of how to start a job run with the Kinesis Data Streams connector dependency.

==
aws emr-serverless start-job-run \
--application-id  \
--execution-role-arn  \
--mode 'STREAMING' \
--job-driver '{
    "sparkSubmit": {
        "entryPoint": "s3://",
        "entryPointArguments": ["s3:///output"],
        "sparkSubmitParameters": "--conf spark.executor.cores=4
                --conf spark.executor.memory=16g 
                --conf spark.driver.cores=4
                --conf spark.driver.memory=16g 
                --conf spark.executor.instances=3
                --jars /usr/share/aws/kinesis/spark-sql-kinesis/lib/spark-streaming-sql-kinesis-connector.jar"
    }
}'
==

To connect to Kinesis Data Streams, you must configure the EMR Serverless application with VPC access and use a VPC endpoint to allow private access. or use a NAT Gateway to get public access. For more information, see Configuring VPC access. You must also make sure that your job runtime role has the necessary read and write permissions to access the required data streams. To learn more about how to configure a job runtime role, see Job runtime roles for Amazon EMR Serverless. For a full list of all of the required permissions, see the spark-sql-kinesis-connector page on GitHub
.
>> Try the above approach and check the lapses

In order to leverage the new API (Structured Streaming)[2], you can leverage the "spark-sql-kinesis-connector" [3].

Please refer to the github project page [4] for more information about the Amazon Kinesis Data Streams Connector for Spark Structured Streaming.

For reference and convenience I'm sharing with you additional resources.
Please note that this resources does not show exactly how to implement the use-case leveraging the spark-sql-kinesis-connector over EMR Serverless however, the similar use-case can provide helpful insights:

* (EMR on EC2) - Spark Structured Streaming Connector for Amazon Kinesis Data Streams using Amazon EMR

https://github.com/aws-samples/spark-kinesis-connector-emr-sample

* (Serverless and Firehose) Run a data processing job on Amazon EMR Serverless with AWS Step Functions

https://aws.amazon.com/blogs/big-data/run-a-data-processing-job-on-amazon-emr-serverless-with-aws-step-functions/

I hope you find the above information clear and helpful.
Should you have any questions or concerns, please feel free to write back.

Have a great day!

====== REFERENCES ======

[1] Spark Streaming - https://spark.apache.org/docs/latest/streaming-programming-guide.html
[2] Structured Streaming - https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html
[3] Apache Spark - https://docs.aws.amazon.com/streams/latest/dev/using-other-services-read-spark.html
[4] https://github.com/awslabs/spark-sql-kinesis-connector

Interaction between EMR Serverless and Kinesis

Add your answer

Relevant content