oracle - 使用 JDBC Source 和 Redis Stream 进行 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 数据库，然后使用 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 流。