spark-redis 实现redis SQL统计

spark-redis是基于jedis实现的redis rdd，可对redis的String, Hash, List, Set and Sorted Set，XStream数据结构进行转换读写，支持将redis数据转换成Dataframes的方式，以Spark SQL进行统计运算，支持官方Redis cluster的集群读取模型，可自动感知Redis分区，亦可根据配置调整分区数。

支持版本：

使用方式，如使用2.4.x版本可引用Maven库：

    
      com.redislabs
      spark-redis_2.11
      2.4.2
    
  

or

    
      com.redislabs
      spark-redis_2.12
      2.4.2
    
  

使用2.3.x版本需要下载源代码：

git clone https://github.com/RedisLabs/spark-redis.git

更改Spark版本号

然后进入到相应的目录进行编译：

cd spark-redis
mvn clean package -DskipTests

编译后，将编译好的包install到本地的maven库

mvn install:install-file -DgroupId=com.redislabs -DartifactId=spark-redis_2.11 -Dversion=3.0.0-SNAPSHOT -Dpackaging=jar -Dfile=.spark-redis_2.11-3.0.0-SNAPSHOT.jar

在对应的POM中加载对应包：

    com.redislabs
    spark-redis_2.11
    3.0.0-SNAPSHOT


    com.alibaba
    fastjson
    1.2.73



    redis.clients
    jedis
    3.4.1
    jar
    compile

编写测试代码：

SparkConf sparkConf = new SparkConf().setAppName(appName).set("spark.redis.host", "127.0.0.1")
        .set("spark.redis.port", "6000")
        .set("spark.redis.auth","123456"); 
SparkSession spark = SparkSession
                    .builder().master("local[4]")
                    .config(sparkConf)
                    .getOrCreate();

Dataset df = spark.createDataframe(Arrays.asList(
  new Person("John", 35),
  new Person("Peter", 40)), Person.class);
df.write()
  .format("org.apache.spark.sql.redis")
  .option("table", "person2")
  .option("key.column", "name")
  .mode(SaveMode.Overwrite)
  .save(); //对redis进行写入操作
//结果:


//redis手动初始化数据 hset person:1 name mike age 30
// hset person:2 name ken age 40
List schemaFields = new ArrayList();
schemaFields.add(new StructField("_id",StringType,false,metadata.empty())); //其 hash key值默认为_id
schemaFields.add(new StructField("name",StringType,true,metadata.empty()));
schemaFields.add(new StructField("age",IntegerType,true,metadata.empty()));
StructType structType=DataTypes.createStructType(schemaFields);
Dataset ds=spark.read()
                    .format("org.apache.spark.sql.redis")
                    .option("keys.pattern", "person:*")
                    .schema(structType).load();

ds.show();
ds.createOrReplaceTempView("persontbl");
Dataset ds2=spark.sql("select * from persontbl");
ds2.show();

//年龄汇总
Dataset ds3=spark.sql("select sum(age) from persontbl");
ds3.show();

探其代码实现，通过redis的scan方法获取到key值后，通过单节点的连接器，一条条的对key的value进行遍历，该操作在大数据量时，可能导致性能问题，需在网络上进行相应的优化。

同样，其String,List, Set and Sorted Set的实现方式亦是如此。