Use Spark on AnalyticDB for MySQL Data Lakehouse Edition (3.0) to access Redis data over an ENI network - AnalyticDB

This topic describes how to use Spark on AnalyticDB for MySQL Data Lakehouse Edition (3.0) to access data in Redis over an elastic network interface (ENI) network.

Prerequisites

An AnalyticDB for MySQL Data Lakehouse Edition (3.0) cluster is created. For more information, see Create a Data Lakehouse Edition (3.0) cluster.
A database account is created.
- If you use your Alibaba Cloud account to access the cluster, you only need to create a privileged account. For more information, see Create a privileged account.
- If you are a RAM user, you must create both a privileged account and a standard account, and then associate the RAM user with the standard account. For more information, see Create a database account and Associate a RAM user with a database account.
A Job-specific resource group is created. For more information, see Create a resource group.
A Redis instance is created in the same region as your AnalyticDB for MySQL Data Lakehouse Edition (3.0) cluster. For more information, see Step 1: Create an instance.
The Redis instance is added to a security group. The security group's inbound and outbound rules must allow access to the Redis port. For more information, see Configure an IP address whitelist and Add a security group rule.
Object Storage Service (OSS) is activated and a bucket is created. For more information, see Activate OSS and Create a bucket.

Procedure

Download the JAR packages required for Spark on AnalyticDB for MySQL to access Redis. For download links, see spark-redis, jedis, and commons-pool2.

Add the dependencies to the dependencies section of the pom.xml file.

      <dependency>
            <groupId>com.redislabs</groupId>
            <artifactId>spark-redis_2.12</artifactId>
            <version>3.1.0</version>
      </dependency>
      <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>3.9.0</version>
      </dependency>
      <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
            <version>2.11.1</version>
      </dependency>

Write the following sample code to access Redis, and then compile and package the code. In this example, the generated JAR package is named redis_test.jar. The sample code is as follows:

package com.aliyun.spark

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._

object SparkOnRedis {

  def main(args: Array[String]): Unit = {
    val redisHost = args(0)
    val redisPort = args(1)
    val redisPassword = args(2)
    var redisTableName = args(3)

   // Configure Redis information in SparkConf.
    val sparkConf = new SparkConf()
      .set("spark.redis.host", redisHost)
      .set("spark.redis.port", redisPort)
      .set("spark.redis.auth", redisPassword)

    val sparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .getOrCreate()

    // Sample data.
    val data =
      Seq(
        Person("John", 30, "60 Wall Street", 150.5),
        Person("Peter", 35, "110 Wall Street", 200.3)
      )

    // Write data using the Dataset API.
    val dfw = sparkSession.createDataFrame(data)
    dfw.write.format("org.apache.spark.sql.redis")
      .option("model", "hash")
      .option("table", redisTableName)
      .save()

    // Read hash values from Redis in the default mode.
    var loadedDf = sparkSession.read.format("org.apache.spark.sql.redis")
      .option("table", redisTableName)
      .load()
      .cache()
    loadedDf.show(10)

    // Set infer.schema to true. Spark retrieves the schema from Redis.
    loadedDf = sparkSession.read.format("org.apache.spark.sql.redis")
      .option("keys.pattern", redisTableName + ":*")
      .option("infer.schema", "true")
      .load()
    loadedDf.show(10)

    // Specify a schema.
    loadedDf = sparkSession.read.format("org.apache.spark.sql.redis")
      .option("keys.pattern", redisTableName + ":*")
      .schema(StructType(Array(
        StructField("name", StringType),
        StructField("age", IntegerType),
        StructField("address", StringType),
        StructField("salary", DoubleType)
      )))
      .load()
    loadedDf.show(10)

    sparkSession.stop()
  }

}

case class Person(name: String, age: Int, address: String, salary: Double)

Upload the JAR packages that you downloaded in Step 1 and the redis_test.jar file to OSS. For more information, see Simple upload.
Go to the Spark JAR development page.
1. Log on to the AnalyticDB for MySQL console.
2. In the upper-left corner of the page, select the region where the cluster resides.
3. In the left-side navigation pane, click Clusters.
4. On the Data Lakehouse Edition tab, click the cluster ID of the target cluster.
5. In the left-side navigation pane, choose Job Development> Spark JAR Development.
Above the editor, select a Job-specific resource group and a job type. This example uses Batch.

In the editor, enter the following job content.

{
    "name": "<redis-example>",
    "file": "oss://<bucket_name>/redis_test.jar",
    "className": "com.aliyun.spark.<SparkOnRedis>",
    "jars": [
      "oss://<bucket_name>/spark-redis_2.12-3.1.0.jar",
      "oss://<bucket_name>/jedis-3.9.0.jar",
      "oss://<bucket_name>/commons-pool2-2.11.1.jar"
    ],
    "args": [
      -- The internal endpoint of the Redis instance. You can find this on the Instance Information page in the Connection Information section.
      "<r-bp1qsslcssr****.redis.rds.aliyuncs.com>",
      -- The port number of the Redis instance. The value is fixed at 6379.
      "6379",
      -- The database account password for the Redis instance.
      "<your_password>",
      -- The table name in the Redis instance.
      "<redis_table_name>"
    ],
    "conf": {
        "spark.adb.eni.enabled": "true",
        "spark.adb.eni.vswitchId": "vsw-bp17jqw3lrrobn6y****",
        "spark.adb.eni.securityGroupId": "sg-bp163uxgt4zandx****",
        "spark.driver.resourceSpec": "small",
        "spark.executor.instances": 1,
        "spark.executor.resourceSpec": "small"
    }
}

The following table describes the parameters.

Parameter	Description
`name`	The name of the Spark job.
`file`	The storage path of the Spark job's main file. The main file can be a JAR package that contains the entry class, or a Python entry script. Note Currently, the main file of a Spark job must be stored in OSS.
`className`	The entry class for the Java or Scala program. This parameter is not required for Python programs.
`args`	The parameters required by your JAR package. Separate multiple parameters with a comma.
`spark.adb.eni.enabled`	Specifies whether to enable access over an ENI network. You must set this to `true` to access a Redis data source.
`spark.adb.eni.vswitchId`	The vSwitch ID. You can find this ID on the Instance Information page of the Redis instance.
`spark.adb.eni.securityGroupId`	The ID of the security group that contains the Redis instance. If you have not configured a security group, see Configure an IP address whitelist.
Other conf parameters	These parameters are similar to standard open-source Spark configurations. Use the `key:value` format and separate multiple parameters with a comma. For more information about conf parameters, see Conf parameters.

Click Run Now.
When the application's status in the application list changes to Succeeded, click Logs in the Actions column to view the data in the Redis table.