如何将流式数据帧写入 PostgreSQL?
Posted
技术标签:
【中文标题】如何将流式数据帧写入 PostgreSQL?【英文标题】:How to write streaming dataframe to PostgreSQL? 【发布时间】:2019-02-18 23:27:39 【问题描述】:我有一个流式数据框,我正在尝试将其写入数据库。有将 rdd 或 df 写入 Postgres 的文档。但是,我无法找到有关如何在结构化流中完成的示例或文档。
我已阅读文档 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)
【问题讨论】:
【参考方案1】:除了你已经拥有的之外,这里真的没什么可做的。 foreachBatch
takes a function (DataFrame, Int) => None
,所以您只需要一个小适配器,其他一切都应该可以正常工作:
def foreach_batch_for_config(config)
def _(df, epoch_id):
postgres_sink(config, df)
return _
view_counts_query = (windowed_view_counts
.writeStream
.outputMode("append")
.foreachBatch(foreach_batch_for_config(some_config))
...,
.start()
.awaitTermination())
虽然说实话,传递ConfigParser
从一开始就是一个奇怪的想法。您可以调整签名并对其进行初始化
def postgres_sink(data_frame, batch_id):
config = configparser.ConfigParser()
...
data_frame.write.jdbc(...)
并保持其余部分不变。这样你就可以直接使用你的函数了:
...
.foreachBatch(postgres_sink)
...
【讨论】:
【参考方案2】:如何使用结构化流完成 Postgres 摄取的示例
class PostgreSqlSink(url: String, user: String, pwd: String) extends org.apache.spark.sql.ForeachWriter[org.apache.spark.sql.Row]
val driver = "org.postgresql.Driver"
var connection: java.sql.Connection = _
var statement: java.sql.PreparedStatement = _
val v_sql = "insert INTO Table (A,B,C) values ( ?, ?, ?)"
def open(partitionId: Long, version: Long): Boolean =
Class.forName(driver)
connection = java.sql.DriverManager.getConnection(url, user, pwd)
connection.setAutoCommit(false)
statement = connection.prepareStatement(v_sql)
true
def process(value: org.apache.spark.sql.Row): Unit =
// ignoring value(0) as this is address
statement.setString(1, value(1).toString)
statement.setString(2, value(2).toString)
statement.setString(3, value(3).toString)
statement.executeUpdate()
def close(errorOrNull: Throwable): Unit =
connection.commit()
connection.close
val url = "jdbc:postgresql://XX.XX.XX.XX:5432/postgres"
val user = "abc"
val pw = "abc@123"
val jdbcWriter = new PostgreSqlSink(url,user,pw)
val writeData = pg_df.writeStream
.foreach(jdbcWriter)
.outputMode("Append")
.trigger(ProcessingTime("30 seconds"))
.option("checkpointLocation", "s3a://bucket/check")
.start()
writeData.awaitTermination
【讨论】:
OP 正在使用 Pyspark。以上是关于如何将流式数据帧写入 PostgreSQL?的主要内容,如果未能解决你的问题,请参考以下文章
FlinkFlink 写入 AnalyticDB MySQL
如何在流式查询中使用外部数据库(postgresql)作为输入?