Mocked Kinesis (Localstack) with PySpark Streaming

Continuing with the same PySpark (ver 2.1.0, Python3.5, etc.) setup explained in an earlier post. In order to connect to the mocked Kinesis stream on Localstack from PySpark use the kinesis_wordcount_asl.py script located in Spark external/ (connector/) folder.

(a) Update value of master in kinesis_wordcount_asl.py

Update value of master(local[n], spark://localhost:7077, etc) in SparkContext in kinesis_wordcount_asl.py:
    sc = SparkContext(appName="PythonStreamingKinesisWordCountAsl",master="local[2]")

(b) Add aSpark compiled jars to Spark Driver/ Executor Classpath

As explained in step (III) of an earlier post, to work with Localstack a few changes were done to the KinesisReceiver.scala onStart() to explicitly set endPoint on kinesis, dynamoDb, cloudWatch clients. Accordingly the compiled aSpark jars with the modifications need to be added to Spark Driver/ Executor classpath.

• For Spark local mode (master="local[n]"): additions to classpath can be exported in the SPARK_CLASSPATH variable.

     export aSPARK_PROJ_HOME="/Downlaod/Location/aSpark"
    export SPARK_CLASSPATH="${aSPARK_PROJ_HOME}/target/original-aSpark_1.0-2.1.0.jar:${aSPARK_PROJ_HOME}/target/scala-2.11/classes:${aSPARK_PROJ_HOME}/target/scala-2.11/jars/*"

•  For Spark Standalone mode: "spark.executor.extraClassPath" needs to be set in either spark-defaults.conf or added as a SparkConf to SparkContext (see (II)(a))

(c) Ensure SPARK_HOME, PYSPARK_PYTHON & PYTHONPATH variables are exported.

(d) Run kinesis_wordcount_asl

    python3.5 ${SPARK_HOME}/external/kinesis-asl/src/main/python/examples/streaming/kinesis_wordcount_asl.py SampleKinesisApplication myFirstStream http://localhost:4566/ us-east-1

• put-records to Localstack Kinesis

    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata abcd"

• Count of the words streamed (put) will show up on the kinesis_wordcount_asl console

 

Spark Streaming with Kinesis mocked on Localstack

In this post we get a Spark streaming application working with AWS Kinesis stream, a mocked version of Kinesis running locally on Localstack. In earlier posts we have explained how to get Localstack running and various AWS services up on Localstack. The client connections to AWS services (Localstack) is done using AWS cli and AWS Java-Sdk v1.

Environment: This set-up continues on a Ubuntu20.04, with Java-8, Maven-3.6x, Docker-24.0x, Python3.5, PySpark/ Spark-2.1.0, Localstack-3.8.1, AWS Java-Sdk-v1 (ver.1.12.778),

Once the Localstack installation is done, steps to follow are:

(I) Start Localstack
    # Start locally
    localstack start

    That should get Localstack should be running on: http://localhost:4566

(II) Check Kinesis services from CLI on Localstack

    # List Streams
    aws --endpoint-url=http://localhost:4566 kinesis list-streams

    # Create Stream
    aws --endpoint-url=http://localhost:4566 kinesis create-stream --stream-name myFirstStream --shard-count 1

    # List Streams
    aws --endpoint-url=http://localhost:4566 kinesis list-streams

    # describe-stream-summary
    aws --endpoint-url=http://localhost:4566 kinesis describe-stream-summary --stream-name myFirstStream

    # Put Record
    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata abcd"
    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata efgh"

(III) Connect to Kinesis from Spark Streaming

• Download & Build a sample aSpark - Java Kinesis application.
• The code is similar to Spark's kinesis-asl from external (connector) module. Except for a few changes to KinesisReceiver.scala onStart() method to explicitly set endPoint on kinesis, dynamoDb, cloudWatch clients. This enables Localstack endPoint url to be plugged into kinesis, dynamoDb & cloudwatch.

    # Build
    mvn install -DskipTests=true -Dcheckstyle.skip

    # Run JavaKinesisWordCountASL with Localstack

• JavaKinesisWordCountASL SampleKinesisApplication myFirstStream http://localhost:4566/
  

• runJavaKinesisWordCountASL.sh script located in sbin/ folder of the aSpark project can be used to run JavaKinesisWordCoundASL from the shell
  

(IV) Add Data to Localstack Kinesis & View Counts on Console
    a) Put Record from cli
    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata abcd"
    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata efgh"

    b) Alternatively Put records from Java Kinesis application
    Download, build & run AmazonKinesisRecordProducerSample.java
    
    c) Now check the output console of JavaKinesisWordCountASL run in step (III) above. Counts of the words streamed from Localstack Kinesis will be displayed on the console.

AWS Lambda on Localstack using Java-SdK-v1

Continuing with Localstack, next is a closer look into the code to deploy and execute AWS Lambda code on Localstack from AWS Java-Sdk-v1. The localstack-lambda-java-sdk-v1 code uses the same structure used in localstack-aws-sdk-examples & fills in for the missing AWS Lambda bit.

The LambdaService class has 3 primary methods - listFunctions(), createFunction() & invokeFunction().  The static AWSLambda client is setup with Mock credentials and pointing to the Localstack endpoint.
 
The main() method first creates the function (createFunction()), if it does not exist.

• It builds a CreateFunctionRequest object with the handler, runtime, role, etc specified
• It also reads the jar file of the Java executable from the resources folder into a FunctionCode object & adds it to the CreateFunctionRequest
• Next a call is made to the AWSLambda client createFunction() with the CreateFunctionRequest which hits the running Localstack instance (Localstack set-up explained earlier).

If all goes well, control returns to main() which invokes the listFunctions() to show details of the created Lambda function (& all others functions existing).

Finally, there is call from main() to invokeFunction() method.

• Which invokes the recently created function with a InvokeRequest object filled with some test values as the payload.
• The response from the invoked function is a InvokeResult object who's payload contains the results of the lambda function computation.

Comments welcome, localstack-lambda-java-sdk-v1 is available to play around!

 

Getting Localstack Up and Running

In continuation to the earlier post on mocks for clouds, this article does a deep dive into getting up & running with Localstack. This is a consolidation of the steps & best practices shared here, here & here. The Localstack set-up is on a Ubuntu-20.04, with Java-8x, Maven-3.8x, Docker-24.0x. 

(I) Set-up

    # Install awscli
     sudo apt-get install awscli

    # Install localstack ver 3.8
        ## Issue1: By default pip pulls in version 4.0, which gives an error:
        ## ERROR: Could not find a version that satisfies the requirement localstack-ext==4.0.0 (from localstack) 

        python3 -m pip install localstack==3.8.1

--

    # Add to /etc/hosts
    127.0.0.1    localhost.localstack.cloud
    127.0.0.1    s3.localhost.localstack.cloud

--

    # Configure AWS from cli
    aws configure
    aws configure set default.region us-east-1
    aws configure set aws_access_key_id test
    aws configure set aws_secret_access_key test

    ## Manually configure AWS
    Add to ~/.aws/config
    endpoint_url = http://localhost:4566

    ## Add mock credentials
    Add to ~/.aws/credentials
    aws_access_key_id = test
    aws_secret_access_key = test

--

    # Download docker images needed by the Lambda function
        ## Issue 2: Do this before hand, Localstack gets stuck
        ## at the download image stage unless it's already available

    ## Pull java:8.al2
    docker pull public.ecr.aws/lambda/java:8.al2

    ## Pull nodejs (required for other nodejs Lambda functions)
    docker pull public.ecr.aws/lambda/nodejs:18

    ## Check images downloaded
    docker image ls

(II) Start Localstack

    # Start locally
    localstack start

    # Start as docker (add '-d' for daemon)
       ## Issue 3: Local directory's mount should be as per sample docker-compose

    docker-compose -f docker-compose-localstack.yaml up

    # Localstack up on URL's
    http://localhost:4566
    http://localhost.localstack.cloud:4566

    # Check Localstack Health
    curl http://localhost:4566/_localstack/info
    curl http://localhost:4566/_localstack/health

(III) AWS services on Localstack from CLI

  (a) S3
    # Create bucket named "test-buck"
    aws --endpoint-url=http://localhost:4566 s3 mb s3://test-buck

    # Copy item to bucket
    aws --endpoint-url=http://localhost:4566 s3 cp a1.txt s3://test-buck

    # List bucket
    aws --endpoint-url=http://localhost:4566 s3 ls s3://test-buck

--

  (b) Sqs
    # Create queue named "test-q"
    aws --endpoint-url=http://localhost:4566 sqs create-queue --queue-name test-q

    # List queues
    aws --endpoint-url=http://localhost:4566 sqs list-queues

    # Get queue attribute
    aws --endpoint-url=http://localhost:4566 sqs get-queue-attributes --queue-url http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/test-q --attribute-names All

--

  (c) Lambda
    aws --endpoint-url=http://localhost:4566 lambda list-functions

    # Create Java function
    aws --endpoint-url=http://localhost:4566 lambda create-function --function-name test-j-div --zip-file fileb://original-java-basic-1.0-SNAPSHOT.jar --handler example.HandlerDivide::handleRequest --runtime java8.al2 --role arn:aws:iam::000000000000:role/lambda-test

    # List functions
    aws --endpoint-url=http://localhost:4566 lambda list-functions

    # Invoke Java function
    aws --endpoint-url=http://localhost:4566 lambda invoke --function-name test-j-div --payload '[200,9]' outputJ.txt

    # Delete function
    aws --endpoint-url=http://localhost:4566 lambda delete-function --function-name test-j-div

(IV) AWS services on Localstack from Java-SDK

    # For S3 & Sqs - localstack-aws-sdk-examples, java sdk

    # For Lambda - localstack-lambda-java-sdk-v1