Un brutto problema con kafka e Spark Streaming su Python
NB Questo NON è lo stesso problema che avevo nel mio primo post su questo sito, tuttavia è lo stesso progetto.
Sto importando alcuni file in PostgreSQL da Kafka usando Spark Streaming. Questi sono i miei passaggi per il progetto:
1- Creazione di uno script per il produttore di kafka (fatto, funziona bene)
2- Creazione di uno script python che legge i file dal produttore di kafka
3- Invio di file a PostgreSQL
Per la connessione tra python e postgreSQL utilizzo psycopg2. Sto anche usando python 3 e java jdk1.8.0_261 e l'integrazione tra kafka e spark streaming funziona bene. Ho kafka 2.12-2.6.0 e spark 3.0.1 e ho aggiunto questi barattoli nella mia directory barattoli Spark:
- 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
Ho dovuto anche scaricare VC ++ per risolvere un altro problema legato anche al mio progetto.
Questo è il mio pezzo di codice python che prende i file dal produttore di kafka e li invia in una tabella di postgreSQL, che ho creato in postgreSQL, in cui ho problemi:
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 è il dataframe spark che ho creato con i file del produttore kafka. process_row è la funzione che inserisce ogni riga del dataframe di streaming nella tabella postgre. Ecco qui:
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
Il problema che ottengo quando eseguo il mio codice si verifica query = satelliteTable.writeStream.outputMode("append").foreachBatch(process_row) \ .option("checkpointLocation", "C:\\Users\\Vito\\Documents\\popo").start()e in breve è il seguente:
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
Il fatto divertente è che lo stesso codice funziona bene sul laptop del mio amico, con spark 3.0.0. Quindi, penso che mi manchino alcuni barattoli o altre cose, perché il codice è corretto.
Qualche idea? Grazie.
Risposte
ti manca questo vaso https://mvnrepository.com/artifact/org.apache.commons/commons-pool2Prova quella versione specificahttps://mvnrepository.com/artifact/org.apache.commons/commons-pool2/2.6.2