Offline Spark consumption example

Sample code

The following Scala program reads log data from a Logstore for a specified time range and saves the output to a file system path.

// TestBatchLoghub.scala

object TestBatchLoghub {
  def main(args: Array[String]): Unit = {
    if (args.length < 6) {
      System.err.println(
        """Usage: TestBatchLoghub <sls project> <sls logstore> <sls endpoint>
          |  <access key id> <access key secret> <output path> <start time> <end time=now>
        """.stripMargin)
      System.exit(1)
    }

    val loghubProject    = args(0)
    val logStore         = args(1)
    val endpoint         = args(2)
    val accessKeyId      = args(3)     // Read from environment variable at submit time
    val accessKeySecret  = args(4)     // Read from environment variable at submit time
    val outputPath       = args(5)
    val startTime        = args(6).toLong

    val sc = new SparkContext(new SparkConf().setAppName("test batch loghub"))
    var rdd: JavaRDD[String] = null

    if (args.length > 7) {
      // Read log data between startTime and endTime
      rdd = LoghubUtils.createRDD(sc, loghubProject, logStore, accessKeyId, accessKeySecret, endpoint, startTime, args(7).toLong)
    } else {
      // Read log data from startTime to now
      rdd = LoghubUtils.createRDD(sc, loghubProject, logStore, accessKeyId, accessKeySecret, endpoint, startTime)
    }

    rdd.saveAsTextFile(outputPath)
  }
}

LoghubUtils.createRDD() returns a JavaRDD[String] where each element is a log entry. Pass seven arguments to read up to the current time, or eight arguments to specify an explicit end time.

For the Maven POM configuration, see aliyun-emapreduce-demo.

Connection parameters

Compile and run

Step 1: Compile the code.

mvn clean package -DskipTests

The compiled JAR is saved to the target/shaded/ directory.

Step 2: Submit the job.

Set the environment variables before submitting:

export ALIBABA_CLOUD_ACCESS_KEY_ID=<access_key_id>
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<access_key_secret>

Then submit the job:

spark-submit \
  --master yarn-cluster \
  --executor-cores 2 \
  --executor-memory 1g \
  --driver-memory 1g \
  --num-executors 2 \
  --class x.x.x.TestBatchLoghub xxx.jar \
  <sls project> <sls logstore> <sls endpoint> \
  $ALIBABA_CLOUD_ACCESS_KEY_ID $ALIBABA_CLOUD_ACCESS_KEY_SECRET \
  <output path> <start time> [<end time>]

Replace x.x.x.TestBatchLoghub with the actual fully qualified class name and xxx.jar with the actual JAR path. Adjust --executor-cores, --executor-memory, and --num-executors based on your data volume and cluster capacity.

Launch spark-sql with the LogHub JAR

spark-sql \
  --jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/* \
  --hiveconf accessKeyId=$ALIBABA_CLOUD_ACCESS_KEY_ID \
  --hiveconf accessKeySecret=$ALIBABA_CLOUD_ACCESS_KEY_SECRET

If your EMR cluster uses Spark 2, replace spark3 with spark2 in the JAR path:

/opt/apps/SPARK-EXTENSION/spark-extension-current/spark2-emrsdk/*

Create a table and query data

CREATE TABLE test_sls
USING loghub
OPTIONS (
  endpoint         = 'cn-hangzhou-intranet.log.aliyuncs.com',
  access.key.id    = '${hiveconf:accessKeyId}',
  access.key.secret= '${hiveconf:accessKeySecret}',
  sls.project      = 'test_project',
  sls.store        = 'test_store',
  startingoffsets  = 'earliest'
);

SELECT * FROM test_sls;

Connection parameters

Use the LogHub JAR in an on-premises development environment

To develop and test locally against Spark 3 (the steps are the same for Spark 2), install the EMR data source JAR to your local Maven repository.

Step 1: Download the JAR from your EMR cluster.

Copy the JAR from the following path on your cluster node to your local machine:

/opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/emr-datasources_shaded_2.12

Step 2: Install the JAR to your local Maven repository.

mvn install:install-file \
  -DgroupId=com.aliyun.emr \
  -DartifactId=emr-datasources_shaded_2.12 \
  -Dversion=3.0.2 \
  -Dpackaging=jar \
  -Dfile=<path-to-downloaded-jar>

Replace <path-to-downloaded-jar> with the local path where you saved the JAR.

Step 3: Add the dependency to your pom.xml.

<dependency>
  <groupId>com.aliyun.emr</groupId>
  <artifactId>emr-datasources_shaded_2.12</artifactId>
  <version>3.0.2</version>
</dependency>