如何使用spark-redis读取spark中的redis地图

Question

我在Redis中有一个普通的scala map (键和值)。现在，我想在我的一个spark-streaming程序中读取该映射，并将其用作广播变量，以便我的从机可以使用该映射来解析键映射。我使用的是spark-redis 2.3.1库，但现在我确定如何阅读它。

在redis表"employee“中映射-

name   |    value
------------------
123         David
124         John
125         Alex

这就是我试图在spark中阅读的方式(不确定这是否正确-请纠正我) --

 val loadedDf = spark.read
  .format("org.apache.spark.sql.redis")
  .schema(
    StructType(Array(
      StructField("name", IntegerType),
      StructField("value", StringType)
    )
  ))
  .option("table", "employee")
  .option("key.column", "name")
  .load()
loadedDf.show() 

上面的代码没有显示任何东西，我得到的输出是空的。

Answer 1

你可以使用下面的代码来完成你的任务，但是你需要利用Spark数据集(case Dataframe to case类)来完成这个任务。下面是一个完整的Redis读写示例。

object DataFrameExample {

  case class employee(name: String, value: Int)

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
          .builder()
          .appName("redis-df")
          .master("local[*]")
          .config("spark.redis.host", "localhost")
          .config("spark.redis.port", "6379")
          .getOrCreate()

    val personSeq = Seq(employee("John", 30), employee("Peter", 45)
    val df = spark.createDataFrame(personSeq)

    df.write
      .format("org.apache.spark.sql.redis")
      .option("table", "person")
      .mode(SaveMode.Overwrite)
      .save()

    val loadedDf = spark.read
                        .format("org.apache.spark.sql.redis")
                        .option("table", "person")
                        .load()
    loadedDf.printSchema()
    loadedDf.show()
  }
}

输出如下

root
 |-- name: string (nullable = true)
 |-- value: integer (nullable = false)

+-----+-----+
| name|value|
+-----+-----+
| John| 30  |
|Peter| 45  |
+-----+-----+

您还可以在Redis文档中查看更多详细信息