使用 JDBC 源和 Redis 流的 Spark Streaming

Question

我正在尝试构建一些技术组合，以在我的工作中实施解决方案。由于我对他们中的大多数人都是新手，所以有时我会被卡住，但可以找到解决我面临的一些问题的方法。现在，这两个对象都在 Spark 上运行，但我似乎无法确定 Streaming 不起作用的原因。

也许是 redis 在写入流端实现其接收器的方式，也许是我试图完成这项工作的方式。我在流式传输上找到的几乎所有示例都与 Spark 示例有关，例如流式传输文本或 TCP，而我在关系数据库上找到的唯一解决方案是基于 kafka connect，我现在无法使用，因为公司没有Kafka 上有 CDC 的 Oracle 选项。

我的场景如下。构建一个 Oracle -> Redis Stream -> MongoDB Spark 应用程序。

我已经根据spark redis 的示例构建了我的代码，并使用示例代码来尝试实现我的案例的解决方案。我每天加载 Oracle 数据并发送到 redis 流，稍后将从流中提取并保存到 Mongo。现在下面的示例只是试图从流中删除并在控制台上显示，但没有显示任何内容。

我尝试过的小“技巧”​​是创建一个 CSV 目录，从中读取，然后从 csv 中获取日期并用于查询 oracle db，然后使用 foreachBatch 命令将 oracle DataFrame 保存到 redis .数据被保存了，但我认为不正确，因为使用示例代码读取流没有收到任何内容。

这些是代码：

** 写入流 **

object SendData extends App {
  Logger.getLogger("org").setLevel(Level.INFO)
  val oracleHost = scala.util.Properties.envOrElse("ORACLE_HOST", "<HOST_IP>")
  val oracleService = scala.util.Properties.envOrElse("ORACLE_SERVICE", "<SERVICE>")
  val oracleUser = scala.util.Properties.envOrElse("ORACLE_USER", "<USER>")
  val oraclePwd = scala.util.Properties.envOrElse("ORACLE_PWD", "<PASSWD>")
  val redisHost = scala.util.Properties.envOrElse("REDIS_HOST", "<REDIS_IP>")
  val redisPort = scala.util.Properties.envOrElse("REDIS_PORT", "6379")
  val oracleUrl = "jdbc:oracle:thin:@//" + oracleHost + "/" + oracleService
  val userSchema = new StructType().add("DTPROCESS", "string")
  val spark = SparkSession
    .builder()
    .appName("Send Data")
    .master("local[*]")
    .config("spark.redis.host", redisHost)
    .config("spark.redis.port", redisPort)
    .getOrCreate()
  val sc = spark.sparkContext
  val sqlContext = new org.apache.spark.sql.SQLContext(sc)
  import sqlContext.implicits._
  val csvDF = spark.readStream.option("header", "true").schema(userSchema).csv("/tmp/checkpoint/*.csv")
  val output = csvDF
    .writeStream
    .outputMode("update")
    .foreachBatch {(df :DataFrame, batchId: Long) => {
      val dtProcess = df.select(col("DTPROCESS")).first.getString(0).take(10)
      val query = s"""
        (SELECT 
            <FIELDS>
        FROM 
            TABLE
        WHERE
          DTPROCESS BETWEEN (TO_TIMESTAMP('$dtProcess 00:00:00.00', 'YYYY-MM-DD HH24:MI:SS.FF') + 1)
          AND (TO_TIMESTAMP('$dtProcess 23:59:59.99', 'YYYY-MM-DD HH24:MI:SS.FF') + 1)
        ) Table
      """
      val df = spark.read
        .format("jdbc")
        .option("url", oracleUrl)
        .option("dbtable", query)
        .option("user", oracleUser)
        .option("password", oraclePwd)
        .option("driver", "oracle.jdbc.driver.OracleDriver")
        .load()
      df.cache()
      if (df.count() > 0) {
        df.write.format("org.apache.spark.sql.redis")
          .option("table", "process")
          .option("key.column", "PRIMARY_KEY")
          .mode(SaveMode.Append)
          .save()
      }
      if ((new DateTime(dtProcess).toLocalDate()).equals(new LocalDate()))
        Seq(dtProcess).toDF("DTPROCESS")
          .coalesce(1)
          .write.format("com.databricks.spark.csv")
          .mode("overwrite")
          .option("header", "true")
          .save("/tmp/checkpoint")
      else {
        val nextDay = new DateTime(dtProcess).plusDays(1)
        Seq(nextDay.toString(DateTimeFormat.forPattern("YYYY-MM-dd"))).toDF("DTPROCESS")
          .coalesce(1)
          .write.format("com.databricks.spark.csv")
          .mode("overwrite")
          .option("header", "true")
          .save("/tmp/checkpoint")
        }
      }}
    .start()
  output.awaitTermination()
}

** 从流中读取 **

object ReceiveData extends App {
  Logger.getLogger("org").setLevel(Level.INFO)
  val mongoPwd = scala.util.Properties.envOrElse("MONGO_PWD", "bpedes")
  val redisHost = scala.util.Properties.envOrElse("REDIS_HOST", "<REDIS_IP>")
  val redisPort = scala.util.Properties.envOrElse("REDIS_PORT", "6379")
  val spark = SparkSession
    .builder()
    .appName("Receive Data")
    .master("local[*]")
    .config("spark.redis.host", redisHost)
    .config("spark.redis.port", redisPort)
    .getOrCreate()
  val processes = spark 
    .readStream
    .format("redis")
    .option("stream.keys", "process")
    .schema(StructType(Array(
      StructField("FIELD_1", StringType),
        StructField("PRIMARY_KEY", StringType),
      StructField("FIELD_3", TimestampType),
      StructField("FIELD_4", LongType),
      StructField("FIELD_5", StringType),
      StructField("FIELD_6", StringType),
      StructField("FIELD_7", StringType),
      StructField("FIELD_8", TimestampType)
    )))
    .load()
  val query = processes
    .writeStream
    .format("console")
    .start()
  query.awaitTermination()
}

Answer 1

此代码将数据帧作为哈希写入 Redis（而不是写入 Redis 流）。

df.write.format("org.apache.spark.sql.redis")
          .option("table", "process")
          .option("key.column", "PRIMARY_KEY")
          .mode(SaveMode.Append)
          .save()

Spark-redis 不支持直接写入 Redis 流。