如何解决使用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流。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。