PythonでのkafkaとSparkStreamingの悪い問題

Oct 29 2020

注意：これは、このサイトへの最初の投稿で発生した問題と同じではありませんが、同じプロジェクトです。

Sparkストリーミングを使用してkafkaからPostgreSQLにいくつかのファイルを取り込んでいます。プロジェクトの手順は次のとおりです。

1- kafkaプロデューサー用のスクリプトの作成（完了、正常に動作します）

2-kafkaプロデューサーからファイルを読み取るPythonスクリプトを作成する

3-PostgreSQLへのファイルの送信

pythonとpostgreSQLの間の接続には、psycopg2を使用します。また、python3とjavajdk1.8.0_261を使用しており、kafkaとsparkストリーミングの統合は正常に機能します。kafka2.12-2.6.0とspark3.0.1があり、これらのjarをSparkjarディレクトリに追加しました。

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 ++もダウンロードする必要がありました。

これは、kafkaプロデューサーからファイルを取得してpostgreSQLのテーブルに送信するPythonコードの一部です。これは、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

面白い事実は、同じコードが私の友人のラップトップで、spark3.0.0で正常に実行されることです。だから、コードが正しいので、私はいくつかの瓶や他のものが欠けていると思います。

何か案が？ありがとう。

回答

Matt Oct 29 2020 at 17:05

あなたはこの瓶を逃していますhttps://mvnrepository.com/artifact/org.apache.commons/commons-pool2その特定のバージョンを試してみてくださいhttps://mvnrepository.com/artifact/org.apache.commons/commons-pool2/2.6.2