java - 如何在 Spark Streaming 中使用 redis

Question

我正在构建一个应用程序，它从 redis 中的列表中读取 json 元素并使用 spark 流式传输它们。这是我写的：

public void readTheStream() throws UnknownHostException, IOException {
        SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("Merge").set("redis.host", "localhost")
                .set("redis.port", "6379");;

        JavaSparkContext ctx = JavaSparkContext.fromSparkContext(SparkContext.getOrCreate(sparkConf));
        JavaStreamingContext context = new JavaStreamingContext(ctx, Durations.seconds(1));
}

如何使用 jssc 对象访问 redis。提前致谢。

Answer 1

这是一个读取myList列表项并将其打印到控制台的示例：

SparkConf sparkConf = new SparkConf().setAppName("MyApp").setMaster("local[*]")
                .set("redis.host", "localhost")
                .set("redis.port", "6379");


JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.milliseconds(1000));

RedisConfig redisConfig = new RedisConfig(new RedisEndpoint(sparkConf));

RedisStreamingContext redisStreamingContext = new RedisStreamingContext(jssc.ssc());
String[] keys = new String[]{"myList"};
RedisInputDStream<Tuple2<String, String>> redisStream =
        redisStreamingContext.createRedisStream(keys, StorageLevel.MEMORY_ONLY(), redisConfig);

redisStream.print();

jssc.start();
jssc.awaitTermination();