java.lang.UnsupportedOperationException：数据源org.apache.spark.sql.redis不支持流式读取

Question

尝试使用 spark 2.3(java) 代码从 Redis 读取数据。 能够从 Redis 读取非流式数据，但在从 Redis 流式读取的情况下无法读取数据，我收到以下错误： 1）当我将格式指定为：

Dataset<Row> RedisData = spark.readStream()
                          .format("org.apache.spark.sql.redis")
                          .option("stream.keys","carsstream")
                          .schema(UserSchema5)
                          .load();

错误是：

java.lang.UnsupportedOperationException: Data source org.apache.spark.sql.redis does not support streamed reading

2) 当我将格式指定为：

Dataset<Row> RedisData = spark.readStream()
                              .format("redis")
                              .option("stream.keys","carsstream")
                              .schema(UserSchema5)
                              .load();

错误是

java.lang.ClassNotFoundException: Failed to find data source: redis. Please find packages at http://spark.apache.org/third-party-projects.html

我指定了Jedis(version 3.1.0)、spark-redis(2.3.1)、spark-core_2.11(version 2.3.0) 的罐子。

任何建议都会有所帮助。

Answer 1

根据您面临的错误以及您尝试实现的代码，我推断您正在使用 Spark 结构化流处理。请参考下面的 sn-ps 供您参考。另外，我分享了指向我的 GitHub 存储库的链接，您可以在其中找到完整的代码。

您需要创建DataFrame/Dataset 而不是Streaming DataFrame/Dataset。 因此，您必须执行以下操作：

val keysPattern = s"${topic}:*"

// SCHEMA FOR STATE DATA CACHED IN REDIS DATA
val redisSchema = StructType(
        List(
            StructField("col1",StringType,true),
            StructField("col2",StringType,true),
            StructField("col3",StringType,true)
        )
    )
val redisDf = spark.read
                    .format("org.apache.spark.sql.redis")
                    .schema(redisSchema)
                    .option("keys.pattern", keysPattern)
                    .load

您可以像这样加入dataframe 和Streaming DataFrame：

// JOIN THE STREAMING DATA WITH REGULAR DATAFRAME
val joinedDf = streamingDf.joinWith(
                      redisDf,
                      trim(col("col1")) === trim(col("u_id")),
                      "left"
                    ).select("_1.*", "_2.*")

regular DataFrame 和 Streaming DataFrame 的连接导致 Streaming DataFrame。要将Streaming DataFrame 写回Redis，您还需要实现foreach writer。这看起来像这样：

// REDIS CONNECTOR - FOREACHWRITER SINK
val redisForeachWriter : RedisForeachWriter = new RedisForeachWriter("localhost","6379", topic)

// PUSH NEW USER DETAILS TO REDIS FOR STATE REFERENCE
val redisSinkQuery = joinedDf
             .select(
                 "col1", "col2", ... , "coln"
             )
             .writeStream
             .outputMode("update")
             .foreach(redisForeachWriter)
             .start

一个示例 RedisForeachWriter 如下所示：

import org.apache.spark.sql.ForeachWriter
import org.apache.spark.sql.Row
import redis.clients.jedis.Jedis
import scala.collection.mutable.HashMap
import scala.collection.JavaConversions._
import scala.collection.JavaConversions
import org.apache.spark.sql.Dataset

class RedisForeachWriter(val host: String, port: String, val topic: String) extends ForeachWriter[Row]{
    // val host: String = p_host
    // val port: String = p_port

    var jedis: Jedis = _

    def connect() = {
        jedis = new Jedis(host, port.toInt)
    }

    override def open(partitionId: Long, version: Long): Boolean = {
        return true
    }

    override def process(record: Row) = {
        val u_id = record.getString(1);

        if( !(u_id == null || u_id.isEmpty())){
          val columns : Array[String] = record.schema.fieldNames

          if(jedis == null){
              connect()
          }

          for(i <- 0 until columns.length){
            if(! ((record.getString(i) == null) || (record.getString(i).isEmpty()) || record.getString(i) == "") )
              jedis.hset(s"${topic}:" + u_id, columns(i), record.getString(i))
          }
        }
    }

    override def close(errorOrNull: Throwable) = {

    }
}

您可以参考我的Github 并提供类似的用例供您参考，并回复以进行澄清。 https://github.com/krohit-scala/MSStreamingStack

编辑：请务必在您的应用程序 POM 中添加这些依赖项。

     <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.11</artifactId>
      <version>2.3.0</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.11</artifactId>
      <version>2.3.0</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-hive_2.11</artifactId>
      <version>2.3.0</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.11</artifactId>
      <version>2.3.0</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
      <version>2.3.0</version>
    </dependency>
    <dependency>
      <groupId>com.redislabs</groupId>
      <artifactId>spark-redis</artifactId>
      <version>2.3.1</version>
    </dependency>