【发布时间】：2021-02-12 01:53:43
【问题描述】：
注意这与我在本网站的第一篇文章中遇到的问题不同，但它是同一个项目。
我正在使用火花流将一些文件从 kafka 摄取到 PostgreSQL 中。这些是我的项目步骤：
1- 为 kafka 生产者创建脚本（完成，工作正常）
2- 创建一个从 kafka producer 读取文件的 python 脚本
3- 发送文件到 PostgreSQL
对于 python 和 postgreSQL 之间的连接，我使用 psycopg2。我也在使用 python 3 和 java jdk1.8.0_261，并且 kafka 和 spark 流之间的集成工作正常。我有 kafka 2.12-2.6.0 和 spark 3.0.1，我在我的 Spark jars 目录中添加了这些 jar：
- postgresql-42.2.18 -spark-streaming-kafka-0-10-assembly_2.12-3.0.1
- spark-token-provider-kafka-0.10_2.12-3.0.1
- kafka-clients-2.6.0
- spark-sql-kafka-0-10-assembly_2.12-3.0.1
我还必须下载 VC++ 才能解决与我的项目相关的另一个问题。
这是我的一段 python 代码，它从 kafka 生产者那里获取文件并将它们发送到我在 postgreSQL 中创建的 postgreSQL 表中，但我遇到了问题：
query = satelliteTable.writeStream.outputMode("append").foreachBatch(process_row) \
.option("checkpointLocation", "C:\\Users\\Vito\\Documents\\popo").start()
print("Starting")
print(query)
query.awaitTermination()
query.stop()
satelliteTable 是我使用来自 kafka 生产者的文件创建的 spark 数据框。 process_row 是将流数据帧的每一行插入 postgre 表的函数。这里是：
def process_row(df, epoch_id):
for row in df.rdd.collect():
cursor1.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)
connection.commit()
pass
我在运行代码时遇到的问题发生在query = satelliteTable.writeStream.outputMode("append").foreachBatch(process_row) \ .option("checkpointLocation", "C:\\Users\\Vito\\Documents\\popo").start()，简而言之如下：
py4j.protocol.Py4JJavaError: An error occurred while calling
z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1
times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0, DESKTOP-
D600TY.homenet.telecomitalia.it, executor driver): java.lang.NoClassDefFoundError:
org/apache/commons/pool2/impl/GenericKeyedObjectPoolConfig
=== Streaming Query ===
Identifier: [id = 599f75a7-5db6-426e-9082-7fbbf5196db9, runId = 67693586-27b1-4ca7-9a44-0f69ad90eafe]
Current Committed Offsets: {}
Current Available Offsets: {KafkaV2[Subscribe[bogi2890.20n]]: {"bogi2890.20n":{"0":68}}}
Current State: ACTIVE
Thread State: RUNNABLE
有趣的事实是，相同的代码在我朋友的笔记本电脑上运行良好，使用 spark 3.0.0。所以，我认为我缺少一些罐子或其他东西，因为代码是正确的。
有什么想法吗？谢谢。
【问题讨论】：
标签： python postgresql apache-kafka spark-streaming rdd