Help Center > > Developer Guide> Developing a DLI Datasource Connection Using a Spark Job> Interconnecting with Redis (By Java)> Detailed Development Description

Detailed Development Description

Updated at: Mar 17, 2020 GMT+08:00

Redis supports only enhanced datasource connections. Only queues in yearly/monthly packages can be used.

Prerequisites

An enhanced datasource connection has been created on the DLI management console and bound to a queue in yearly/monthly packages.

Code Implementation

  1. Dependencies related to import
    1
    2
    3
    4
    5
    6
    7
    8
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.sql.*;
    import org.apache.spark.sql.types.DataTypes;
    import org.apache.spark.sql.types.StructField;
    import org.apache.spark.sql.types.StructType;
    import java.util.*;
    
  2. Create a session
    1
    2
    3
    4
    5
    6
    7
    8
    SparkConf sparkConf = new SparkConf();
    sparkConf.setAppName("datasource-redis")
            .set("spark.redis.host", "192.168.4.199")
            .set("spark.redis.port", "6379")
            .set("spark.redis.auth", "******")
            .set("spark.driver.allowMultipleContexts","true");
    JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
    SQLContext sqlContext = new SQLContext(javaSparkContext);
    

Connecting to Datasources Through DataFrame APIs

  1. Reads JSON data as DataFrame.
    1
    2
    3
    4
    JavaRDD<String> javaRDD = javaSparkContext.parallelize(Arrays.asList(
            "{\"id\":\"1\",\"name\":\"zhangsan\",\"age\":\"18\"}",
            "{\"id\":\"2\",\"name\":\"lisi\",\"age\":\"21\"}"));
    Dataset dataFrame = sqlContext.read().json(javaRDD);
    
  2. Construct the Redis connection configuration parameters.
    1
    2
    3
    Map map = new HashMap<String, String>();
    map.put("table","person");
    map.put("key.column","id");
    
  3. Save data to Redis
    1
    dataFrame.write().format("redis").options(map).mode(SaveMode.Overwrite).save();
    
  4. Read data from Redis
    1
    sqlContext.read().format("redis").options(map).load().show();
    
  5. Operation result

Submitting a Spark Job

  1. Upload the Java code file to DLI. For details about console operations, see the Data Lake Insight User Guide. For API references, see Uploading a Resource Package in the Data Lake Insight API Reference.
  2. In the Spark job editor, select the corresponding dependency and execute the Spark job. For details about console operations, see the Data Lake Insight User Guide. For API references, see Creating a Session (Recommended) and Creating a Batch Processing Job in the Data Lake Insight API Reference.
    • When submitting a job, you need to specify a dependency module named sys.datasource.redis.
    • For details about how to submit a job on the console, see Table 6-Dependency Resources parameter description in the Data Lake Insight User Guide.
    • For details about how to submit a job through an API, see the modules parameter in Table 2-Request parameter description of Creating a Session and Creating a Batch Processing Job in the Data Lake Insight API Reference.

Did you find this page helpful?

Submit successfully!

Thank you for your feedback. Your feedback helps make our documentation better.

Failed to submit the feedback. Please try again later.

Which of the following issues have you encountered?







Please complete at least one feedback item.

Content most length 200 character

Content is empty.

OK Cancel