spark读取redis数据（交互式，scala单机版，java单机版）

Posted 2022-11-13 靖-Drei

• 交互式

第一步：向redis中添加数据

第二步：将jedis jar包放入~/lib目录下，开启spark服务

第三步：通过spark-shell读取redis数据，并做相应处理

• scala单机版

package com.test

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

import redis.clients.jedis.Jedis

object RedisClient 
  def main(args: Array[String]) 

    val conf = new SparkConf()
    conf.setAppName("wow,my first spark app")
    conf.setMaster("local")
    val sc = new SparkContext(conf)
    var jd = new Jedis("172.171.51.154", 6379)
    var str = jd.get("chengshi")
    var strList = str.split(",")
    val a = sc.parallelize(strList, 3)
    val b = a.keyBy(_.length)
    b.collect().foreach(s => println(s._1 + ":" + s._2))

    sc.stop()
  

输出结果

• java单机版

package com.dt.spark.SparkApps.cores;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Arrays;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import redis.clients.jedis.Jedis;
import scala.Tuple2;

public class Redis 

    public static void main(String[] args) throws FileNotFoundException, IOException 

        SparkConf conf = new SparkConf().setAppName("Spark Read From Redis written by Java").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        Jedis jedis = new Jedis("172.171.51.154", 6379);

        String s = jedis.get("chengshi");
        String[] strList = s.split(",");

        JavaRDD<String> ch = sc.parallelize(Arrays.asList(strList), 4);

        JavaPairRDD<String, Integer> jp = ch.mapToPair(new PairFunction<String, String, Integer>() 

            public Tuple2<String, Integer> call(String word) throws Exception 
                return new Tuple2<String, Integer>(word, word.length());
            
        );
        jp.foreach(new VoidFunction<Tuple2<String, Integer>>() 

            public void call(Tuple2<String, Integer> pairs) throws Exception 
                System.out.println(pairs._1() + ":" + pairs._2());
            
        );
        sc.close();

    

输出结果