尝试使用 Kafka 和 pyspark 在 postgreSQL 中从 spark 编写流数据帧答案

【问题标题】：Trying to write a streaming dataframe from spark in postgreSQL with Kafka and pyspark尝试使用 Kafka 和 pyspark 在 postgreSQL 中从 spark 编写流数据帧
【发布时间】：2021-02-09 21:55:34
【问题描述】：

我一直在本网站的各个方面搜索此问题，但没有找到任何解决方案。我编写了一个 java 类，它在 Kafka 中创建一个生产者并发送一些文件，它工作正常。然后，我想编写一个 python 脚本来读取这些文件并将它们放入 postgreSQL 的数据库中。

每个文件（每个文件是一个有很多列的数据集）在kafka消费者中成为一个主题，文件的每一行成为相关主题中的一条消息。

这是我在 python 中根据流数据创建的 spark 数据框：

 list = df.select("fileName", "Satellite_PRN_number", "date", "time", "Crs", "Delta_n", "m0", "Cuc",
                 "e_Eccentricity",
                 "Cus",
                 "sqrt_A", "Toe_Time_of_Ephemeris", "Cic", "OMEGA_maiusc", "cis", "i0", "Crc", "omega",
                 "omega_dot",
                 "idot")

这是我的 python 函数，它应该在我的 postgreSQL 表中插入每一行。我使用 psycopg2 在 python 和 postgre 之间创建连接，并使用“self.cursor.execute”来编写查询。

def process_row(self, row):
  self.cursor.execute(
  'INSERT INTO satellite(fileName,Satellite_PRN_number, date, time,Crs,Delta_n, m0, 
  Cuc,e_Eccentricity,Cus,'
  'sqrt_A, Toe_Time_of_Ephemeris, Cic, OMEGA_maiusc, cis, i0, Crc, omega, omega_dot, idot) VALUES 
  (%s,%s,%s,'
  '%s,%s,%s, %s, %s, %s, %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)',
  (row.fileName, row.Satellite_PRN_number, row.date, row.time, row.Crs, row.Delta_n, row.m0, row.Cuc,
  row.e_Eccentricity,
  row.Cus, row.sqrt_A, row.Toe_Time_of_Ephemeris, row.Cic, row.OMEGA_maiusc, row.cis, row.i0, 
  row.Crc,
  row.omega,
  row.omega_dot, row.idot))
  self.connection.commit()

最后，我使用上面的这个方法通过以下命令在 postgreSQL 中填充我的表：

query = list.writeStream.outputMode("append").foreachBatch(process_row)\ 
        .option("checkpointLocation", "C:\\Users\\Admin\\AppData\\Local\\Temp").start()

我收到以下错误：AttributeError: 'DataFrame' object has no attribute 'cursor'。

我认为问题出在 row.fileName 等...或方法“process_row”中。我不完全了解如何管理方法“process_row”以传递流数据帧的每一行来填充 posteSQL 表。

谁能帮帮我？谢谢。

【问题讨论】：

标签： python postgresql apache-spark apache-kafka spark-streaming-kafka

【解决方案1】：

您的 foreachBatch 签名似乎不正确。应该是这样的：

def foreach_batch_function(df, epoch_id):
    # Transform and write batchDF
    pass
  
streamingDF.writeStream.foreachBatch(foreach_batch_function).start()

如您所见，forEachBatch 函数的第一个参数是一个 DataFrame，而不是您所期望的 psycopg2 类的实例。 ForEachBatch 将有一个 DataFrame，它本身将包含当前微批处理中的所有行，而不仅仅是一行。

因此，您可以尝试在该函数中声明您的 postgreSQL 连接的实例以进一步使用它，或者您可以尝试这种方法：

我会为你的 postgreSQL 数据库创建一个基于 hive jdbc 源的表，如下所示：

CREATE TABLE jdbcTable
USING org.apache.spark.sql.jdbc
OPTIONS (
  url "jdbc:postgresql:dbserver",
  dbtable "schema.tablename",
  user 'username',
  password 'password'
)

这将使您能够像这样使用您的 forEachBatch 函数：

def foreach_batch_function(df, epoch_id):
    # Transform and write batchDF
    df.write.insertInto("jdbcTable")

希望对你有帮助

【讨论】：