我有一个流式数据框,想将其写入数据库。有关如何将RDD或DF写入Postgres的文档,但我找不到在Structured Streaming中执行此操作的示例或文档。
我已阅读文档https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#foreachbatch,但我无法理解在哪里创建JDBC连接以及如何将其写入数据库。
我已阅读文档https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#foreachbatch,但我无法理解在哪里创建JDBC连接以及如何将其写入数据库。
def foreach_batch_function(df, epoch_id):
# what goes in here?
pass
view_counts_query = windowed_view_counts.writeStream \
.outputMode("append") \
.foreachBatch(foreach_batch_function)
.option("truncate", "false") \
.trigger(processingTime="5 seconds") \
.start() \
.awaitTermination()
这个函数接受一个普通的数据帧,并将其写入一个postgres表中。
def postgres_sink(config, data_frame):
config.read('/src/config/config.ini')
dbname = config.get('dbauth', 'dbname')
dbuser = config.get('dbauth', 'user')
dbpass = config.get('dbauth', 'password')
dbhost = config.get('dbauth', 'host')
dbport = config.get('dbauth', 'port')
url = "jdbc:postgresql://"+dbhost+":"+dbport+"/"+dbname
properties = {
"driver": "org.postgresql.Driver",
"user": dbuser,
"password": dbpass
}
data_frame.write.jdbc(url=url, table="metrics", mode="append",
properties=properties)