使用JDBC Source和Redis Stream进行Spark流式传输

问题描述

我正在尝试构建一些技术组合以在我的工作中实施解决方案。由于我是大多数人的新手，有时我会陷入困境，但是可以找到解决我所面临的一些问题的方法。现在，两个对象都在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","6379")
  val spark = SparkSession
    .builder()
    .appName("Receive Data")
    .master("local[*]")
    .config("spark.redis.host",redisPort)
    .getOrCreate()
  val processes = spark 
    .readStream
    .format("redis")
    .option("stream.keys","process")
    .schema(StructType(Array(
      StructField("FIELD_1",StringType),StructField("PRIMARY_KEY",StructField("FIELD_3",TimestampType),StructField("FIELD_4",LongType),StructField("FIELD_5",StructField("FIELD_6",StructField("FIELD_7",StructField("FIELD_8",TimestampType)
    )))
    .load()
  val query = processes
    .writeStream
    .format("console")
    .start()
  query.awaitTermination()
}

解决方法

此代码将数据帧作为哈希写入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流。