A library for reading data from Amzon S3 with optimised listing using Amazon SQS using Spark SQL Streaming ( or Structured streaming.).

Using SBT:

libraryDependencies += "com.qubole" %% "spark-sql-streaming-sqs_{{site.SCALA_BINARY_VERSION}}" % "{{site.PROJECT_VERSION}}"

Using Maven:

<dependency>
    <groupId>com.qubole</groupId>
    <artifactId>spark-sql-streaming-sqs_{{site.SCALA_BINARY_VERSION}}</artifactId>
    <version>{{site.PROJECT_VERSION}}</version>
</dependency>

This library can also be added to Spark jobs launched through spark-shell or spark-submit by using the --packages command line option. For example, to include it when starting the spark shell:

$ bin/spark-shell --packages com.qubole:spark-sql-streaming-sqs_{{site.SCALA_BINARY_VERSION}}:{{site.PROJECT_VERSION}}

Unlike using --jars, using --packages ensures that this library and its dependencies will be added to the classpath. The --packages argument can also be used with bin/spark-submit.

This library is compiled for Scala 2.12 only, and intends to support Spark 2.4.0 onwards.

S3-SQS Connector is built using Apache Maven](http://maven.apache.org/).

To build S3-SQS connector, clone this repository and run:

mvn -DskipTests clean package

This will create target/spark-sql-streaming-sqs_2.11-0.5.1.jar file which contains s3-sqs connector code and associated dependencies. Make sure the Scala and Java versions correspond to those required by your Spark cluster. We have tested it with Java 7/8, Scala 2.11 and Spark version 2.4.0.

The configuration is obtained from parameters.

An example to create a SQL stream which uses Amazon SQS to list files on S3,

    val inputDf = sparkSession
                      .readStream
                      .format("s3-sqs")
                      .schema(schema)
                      .option("sqsUrl", queueUrl)
                      .option("region", awsRegion)
                      .option("fileFormat", "json")
                      .option("sqsFetchIntervalMilliSeconds", "2000")
                      .option("useInstanceProfileCredentials", "true")
                      .option("sqsLongPollingWaitTimeSeconds", "5")
                      .load()