AnalyticDB:Access a Redis data source

Step 1: Download the required JAR packages

Download the following JAR packages from Maven Central:

• spark-redis_2.12 3.1.0

• jedis 3.9.0

• commons-pool2 2.11.1

Step 2: Add Maven dependencies

Add the following dependencies to your pom.xml:

<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>

Step 3: Write and package the Scala program

The following example writes data to Tair and reads it back. Package the program as redis_test.jar.

Write data

The example writes two Person records to Tair as hash entries using the org.apache.spark.sql.redis connector.

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 the Tair (Redis OSS-compatible) connection
    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()

    // Write sample data to Tair as hash entries
    val data = Seq(
      Person("John", 30, "60 Wall Street", 150.5),
      Person("Peter", 35, "110 Wall Street", 200.3)
    )
    val dfw = sparkSession.createDataFrame(data)
    dfw.write.format("org.apache.spark.sql.redis")
      .option("model", "hash")
      .option("table", redisTableName)
      .save()

Read data: pattern 1 — read by table name (recommended)

Use this pattern when the data was written by spark-redis. The schema is inferred from stored metadata, and column types are preserved.

    // Read back a DataFrame previously written by spark-redis.
    // The schema is inferred from the stored metadata.
    var loadedDf = sparkSession.read.format("org.apache.spark.sql.redis")
      .option("table", redisTableName)
      .load()
      .cache()
    loadedDf.show(10)

Read data: pattern 2 — read by key pattern with schema inference

Use this pattern when the hash keys were written externally (not by spark-redis). All columns are returned as strings.

    // Read existing hash keys using a key pattern, with schema inference.
    // All columns are returned as strings.
    loadedDf = sparkSession.read.format("org.apache.spark.sql.redis")
      .option("keys.pattern", redisTableName + ":*")
      .option("infer.schema", "true")
      .load()
    loadedDf.show(10)

Read data: pattern 3 — read by key pattern with explicit schema

Use this pattern when you need specific column types for externally written hash keys.

    // Read existing hash keys with an explicit schema.
    // Use this when you need specific column types.
    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)

Step 4: Upload JAR packages to OSS

Upload all four JAR packages to your OSS bucket using simple upload:

• spark-redis_2.12-3.1.0.jar

• jedis-3.9.0.jar

• commons-pool2-2.11.1.jar

• redis_test.jar (your packaged program)

Step 5: Open Spark JAR Development

1. Log on to the AnalyticDB for MySQL console.

2. In the upper-left corner, select the region where your cluster resides.

3. In the left-side navigation pane, click Clusters.

4. On the Data Lakehouse Edition (V3.0) tab, find your cluster and click its cluster ID.

5. In the left-side navigation pane, choose Job Development > Spark JAR Development.

Step 6: Configure and run the Spark job

1. Select a job resource group and set the job type to batch.

2. Enter the following job configuration in the Spark editor. Replace the placeholders with your actual values.

   {
       "name": "<job-name>",
       "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": [
         "<internal-endpoint>",
         "6379",
         "<password>",
         "<table-name>"
       ],
       "conf": {
           "spark.adb.eni.enabled": "true",
           "spark.adb.eni.vswitchId": "<vswitch-id>",
           "spark.adb.eni.securityGroupId": "<security-group-id>",
           "spark.driver.resourceSpec": "small",
           "spark.executor.instances": 1,
           "spark.executor.resourceSpec": "small"
       }
   }

   Job configuration parameters:

   Parameter	Description	Where to find the value
   name	Name of the Spark job.	Choose any descriptive name.
   file	OSS path of the main JAR file. Must be stored in OSS.	The path where you uploaded redis_test.jar.
   className	Entry class of the Java or Scala program. Not required for Python programs.	com.aliyun.spark.SparkOnRedis
   args	Runtime arguments passed to the JAR, in the order the program expects.	See the argument descriptions below.
   spark.adb.eni.enabled	Enables ENI for the Spark job. Must be true to access Tair.	Fixed value: true
   spark.adb.eni.vswitchId	The vSwitch ID of the Tair instance's VPC.	On the Tair Instance Information page, find the vSwitch ID in the network configuration section.
   spark.adb.eni.securityGroupId	The ID of the ECS security group added to the Tair instance as a whitelist.	On the Tair Instance Information page, find the security group ID. For setup instructions, see the "Method 2: Add ECS security groups as whitelists" section of Configure whitelists.
   conf	Additional Spark configuration parameters in key:value format. Separate multiple parameters with commas. See Conf configuration parameters.	—

   Arguments (args):

   Position	Value	Description
   0	<internal-endpoint>	Internal endpoint of the Tair instance (for example, r-bp1qsslcssr****.redis.rds.aliyuncs.com). Find it in the Connection Information section of the Tair Instance Information page.
   1	6379	Port number of the Tair instance.
   2	<password>	Password for the Tair database account.
   3	<table-name>	Name of the table in the Tair instance.

3. Click Run Now.

Step 7: Verify the results

After the Spark application state changes to Completed, click Log in the Actions column on the Applications tab to view the output data from the Tair table.