ОШИБКА SparkContext: не удалось добавить spark-streaming-kafka-0-10_2.13-3.5.2.jar. ⇐ Python

[Anonymous]

Код: Выделить всё

ERROR SparkContext: Failed to add home/areaapache/software/spark-3.5.2-bin-hadoop3/jars/spark-streaming-kafka-0-10_2.13-3.5.2.jar \
to Spark environment

Код: Выделить всё

import logging
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, explode, to_date, first, last, max, min, avg,     sum, lit
from pyspark.sql.types import StructType, StructField, ArrayType, StringType, DoubleType, LongType,   IntegerType
import time
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

kafka_topic_name = "finnhub_data"
kafka_bootstrap_servers = "localhost:9092"

mysql_host_name = "localhost"
mysql_port_no = "3306"
mysql_database_name = "finnhub_processed"
mysql_driver_class = "com.mysql.cj.jdbc.Driver"   # Cập nhật driver class
mysql_table_name = "processed_trade_data"
mysql_user_name = "tuanle"
mysql_password = "123456"
mysql_jdbc_url = f"jdbc:mysql://{mysql_host_name}:{mysql_port_no}/{mysql_database_name}"

cassandra_host_name = "localhost"
cassandra_port_no = "9042"
cassandra_keyspace_name = "finnhub_data"
cassandra_table_name = "raw_trade_data"

spark_jar = "file:///home/tuanle/areaapache/software/spark-3.5.2-bin-hadoop3/jars"
def save_to_cassandra(df, epoch_id):
logger.info(f"Saving batch {epoch_id} to Cassandra")

df_to_save = df.select(
col("symbol"),
(col("trade_time") / 1000).cast("timestamp").alias("trade_time"),
col("price"),
col("volume"),
col("conditions"),
col("company")
)

df_to_save.write \
.format("org.apache.spark.sql.cassandra") \
.mode("append") \
.options(table=cassandra_table_name, keyspace=cassandra_keyspace_name) \
.save()

logger.info(f"Batch {epoch_id} saved to Cassandra successfully")

def process_and_save_to_mysql(spark):
logger.info("Starting to process data from Cassandra and save to MySQL")

df_cassandra = spark.read \
.format("org.apache.spark.sql.cassandra") \
.options(table=cassandra_table_name, keyspace=cassandra_keyspace_name) \
.load()

df_processed = df_cassandra \
.withColumn("trade_date", to_date(col("trade_time"))) \
.groupBy("symbol", "trade_date") \
.agg(
first("price").alias("open_price"),
last("price").alias("close_price"),
max("price").alias("high_price"),
min("price").alias("low_price"),
avg("price").alias("avg_price"),
sum("volume").alias("total_volume")
) \
.withColumn("processed_time", lit(time.strftime("%Y-%m-%d %H:%M:%S")))

df_processed.write \
.jdbc(url=mysql_jdbc_url,
table=mysql_table_name,
mode="append",
properties={
"user": mysql_user_name,
"password": mysql_password,
"driver":  mysql_driver_class
})

logger.info("Data processed and saved to MySQL successfully")

if __name__ == "__main__":
logger.info("Data Processing Application Started ...")

# Khởi tạo Spark Session và định nghĩa cấu hình cần thiết
spark = SparkSession.builder \
.appName("PySpark Structured Streaming with Kafka, Cassandra, and MySQL") \
.config("spark.streaming.stopGracefullyOnShutdown", True) \
.config("spark.jars", f"{spark_jar}/jsr305-3.0.0.jar,{spark_jar}/spark-cassandra-connector_2.13-3.5.1.jar,{spark_jar}/spark-sql-kafka-0-10_2.13-3.5.2.jar,{spark_jar}/spark-streaming-kafka-0-10_2.13-3.5.2.jar,file:///usr/share/java/mysql-connector-java.jar,{spark_jar}/kafka-clients-3.8.0.jar") \
.config("spark.sql.shuffle.partitions", 4) \
.config("spark.cassandra.connection.host", cassandra_host_name) \
.config("spark.cassandra.connection.port", cassandra_port_no) \
.config("spark.sql.mysql.host", mysql_host_name) \
.config("spark.sql.mysql.port", mysql_port_no) \
.config("spark.cassandra.connection.keep_alive_ms", "60000") \
.master("local[4]") \
.getOrCreate()

spark.sparkContext.setLogLevel("ERROR")

logger.info("Spark Session initialized successfully")

finnhub_df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
.option("subscribe", kafka_topic_name) \
.option("startingOffsets", "latest") \
.load()

logger.info("Printing Schema of finnhub_df:")
finnhub_df.printSchema()

finnhub_schema = StructType([
StructField("data", ArrayType(StructType([
StructField("c", ArrayType(StringType())),
StructField("p", DoubleType()),
StructField("s", StringType()),
StructField("t", LongType()),
StructField("v", IntegerType()),
StructField("company", StringType())
]))),
StructField("type", StringType())
])

finnhub_df1 = finnhub_df.selectExpr("CAST(value AS STRING)", "timestamp")
finnhub_df2 = finnhub_df1.select(from_json(col("value"), finnhub_schema).alias("finnhub"), "timestamp")
finnhub_df3 = finnhub_df2.select("finnhub.data", "timestamp")
finnhub_df4 = finnhub_df3.select(explode("data").alias("trade_data"), "timestamp")
finnhub_df5 = finnhub_df4.select(
col("trade_data.s").alias("symbol"),
col("trade_data.t").alias("trade_time"),
col("trade_data.p").alias("price"),
col("trade_data.v").alias("volume"),
col("trade_data.c").alias("conditions"),
col("trade_data.company").alias("company"),
col("timestamp")
)

query_cassandra = finnhub_df5 \
.writeStream \
.trigger(processingTime='15 seconds') \
.outputMode("append") \
.foreachBatch(save_to_cassandra) \
.start()

try:
while True:
time.sleep(300)  # Wait for 5 minutes
process_and_save_to_mysql(spark)
except KeyboardInterrupt:
logger.info("Application interrupted. Stopping streams...")
query_cassandra.stop()
spark.stop()
logger.info("Application stopped successfully")
except Exception as e:
logger.error(f"An error occurred: {str(e)}")
query_cassandra.stop()
spark.stop()
logger.info("Application stopped due to an error")

Код: Выделить всё

File "/home/tuanle/apachearea/software/spark-3.5.2-bin-hadoop3/python/lib/py4j-0.10.9.7-src.zip/py4j/protocol.py", line 326, \
in get_return_value py4j.protocol.Py4JJavaError: An error occurred while calling o47.load.  : \
java.lang.NoClassDefFoundError: scala/$less$colon$less \
at org.apache.spark.sql.kafka010.KafkaSourceProvider.org$apache$spark$sql$kafka010$KafkaSourceProvider$$validateStreamOptions(KafkaSourceProvider.scala:338) \
at org.apache.spark.sql.kafka010.KafkaSourceProvider.sourceSchema(KafkaSourceProvider.scala:71) \
at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:233) \
at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:118) \
at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:118) \
at org.apache.spark.sql.execution.streaming.StreamingRelation$.apply(StreamingRelation.scala:36) \
at org.apache.spark.sql.streaming.DataStreamReader.loadInternal(DataStreamReader.scala:169) \
at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:145) \
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) \
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) \
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) \
at java.lang.reflect.Method.invoke(Method.java:498) \
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) \
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374) \
at py4j.Gateway.invoke(Gateway.java:282) \
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) \
at py4j.commands.CallCommand.execute(CallCommand.java:79) \
at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182) \
at py4j.ClientServerConnection.run(ClientServerConnection.java:106) \
at java.lang.Thread.run(Thread.java:750) \
Caused by: java.lang.ClassNotFoundException: scala.$less$colon$less \
at java.net.URLClassLoader.findClass(URLClassLoader.java:387) \
at java.lang.ClassLoader.loadClass(ClassLoader.java:418) \
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352) \
at java.lang.ClassLoader.loadClass(ClassLoader.java:351) \
... 20 more
2024-09-26 22:42:24,971 - INFO - Closing down clientserver connection`

Я не знаю, почему я уже загрузил все необходимые файлы jar, соответствующие версии.
Затем сохраните их в папке jars, но Spark все равно не может их найти. .

Подробнее здесь: https://stackoverflow.com/questions/790 ... -3-5-2-jar