我正在尝试构建一些技术组合，以便在我的工作中实现一个解决方案。因为我对他们中的大多数人都是新手，所以有时我会陷入困境，但我能找到解决我所面临的一些问题的方法。现在，这两个对象都在spark上运行，但我似乎无法确定流媒体为什么不起作用。
也许是redis在写流端实现其sink的方式，也许是我正在尝试的工作方式。我在流媒体上找到的几乎所有示例都与spark示例相关，比如流文本或tcp，而我在关系数据库上找到的唯一解决方案是基于kafka connect的，我现在不能使用它，因为该公司没有oracle在kafka上选择cdc。
我的设想如下。构建oracle->redis stream->mongodb spark应用程序。
我已经基于spark redis的示例构建了我的代码，并使用示例代码尝试实现一个针对我的案例的解决方案。我每天加载oracle数据并发送到redis流，稍后将从流中提取并保存到mongo。现在，下面的示例只是试图从流中删除并在控制台上显示，但没有显示任何内容。
我尝试过的一个小技巧是创建一个csv目录，从中读取，然后从csv中获取日期并用于查询oracle数据库，然后使用foreachbatch命令将oracleDataframe保存在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()
}