Can I get your github URL where I can download this codes?

Where is the source of the data set that you got from ? is that from UCI MLR or kaggle or Any website or you scraped ??

C:\Windows\System32>spark-shell Setting default log level to "WARN". To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel). 24/09/26 11:27:59 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable 24/09/26 11:27:59 ERROR Main: Failed to initialize Spark session. java.lang.UnsupportedOperationException: getSubject is supported only if a security manager is allowed

Hi , really liked the video,can you send the dataset please

I did not receive an email sending the OTP code when registering an datamaking account :(((

hi sir, i am getting this error, can you give me the solution? Failed to get schema version. Underlying cause: java.sql.SQLNonTransientConnectionException : Could not create connection to database server. Attempted reconnect 3 times. Giving up. SQL Error code: 0 Use --verbose for detailed stacktrace. *** schemaTool failed *** --verbose org.apache.hadoop.hive.metastore.HiveMetaException: Failed to get schema version. Caused by: java.sql.SQLNonTransientConnectionException: Could not create connection to database server. Attempted reconnect 3 times. Giving up. Caused by: com.mysql.cj.exceptions.CJException: Public Key Retrieval is not allowed Caused by: com.mysql.cj.exceptions.UnableToConnectException: Public Key Retrieval is not allowed *** schemaTool failed ***

Could you please share the source code.. it would be better to follow the same steps and understand in a better way. @datamaking

what is purpose of this if there is no documentation ?

I need vm

For those who don.t whan't the anaconda environment by default each time you open a terminal run this 'conda config --set auto_activate_base false', this will let you open terminals as usual and invoke conda only when desired

Is it necessary to download in virtual machine?

for all who gets error "Microsoft ODBC Driver 18 for SQL Server : Client unable to establish connection." after executing "/opt/mssql-tools/bin/sqlcmd...", just add "-C" at the end of the line.

​ @DataMaking #MAIN_SCRIPT from configparser import ConfigParser from pyspark.sql import SparkSession from pyspark.sql.functions import col, lit, from_json from pyspark.sql.types import StringType, StructType import time # Configuration config_file = 'cred.conf' config = ConfigParser() config.read(config_file) # Kafka Configuration kafka_host_name = config.get("kafka", "host") kafka_port_no = config.get("kafka", "port_no") kafka_bootstrap_server = f"{kafka_host_name}:{kafka_port_no}" input_topic_name = config.get("kafka", "input_topic_name") output_topic_name = config.get("kafka", "output_topic_name") # MySQL Configuration mysql_host_name = config.get('mysql', 'host') mysql_port_no = config.get('mysql', 'port_no') mysql_user_name = config.get('mysql', 'username') mysql_password = config.get('mysql', 'password') mysql_database_name = config.get('mysql', 'db_name') mysql_driver = config.get('mysql', 'driver') mysql_jdbc_url = f"jdbc:mysql://{mysql_host_name}:{mysql_port_no}/{mysql_database_name}" # Cassandra Configuration cassandra_host_name = config.get("cassandra", "host") cassandra_port = config.get("cassandra", "port_no") cassandra_keyspace = config.get("cassandra", "keyspace") cassandra_table_name = config.get("cassandra", "table_name") cassandra_username = config.get("cassandra", "username") cassandra_password = config.get("cassandra", "password") def save_to_cassandra(current_df, epoch_id): print(f"Saving batch {epoch_id} to Cassandra") current_df.write \ .format('org.apache.spark.sql.cassandra') \ .mode('append') \ .options(table=cassandra_table_name, keyspace=cassandra_keyspace) \ .save() def save_to_mysql(current_df, epoch_id): db_credentials = {"user": mysql_user_name, "password": mysql_password, "driver": mysql_driver} processed_at = time.strftime("%Y-%m-%d %H:%M:%S") current_final_df = current_df \ .withColumn("processed_at", lit(processed_at)) \ .withColumn("batch_id", lit(epoch_id)) print(f"Saving batch {epoch_id} to MySQL") current_final_df.write.jdbc(mysql_jdbc_url, "table_name", mode="append", properties=db_credentials) def create_spark_session(): return SparkSession.builder \ .appName("data-processing") \ .master("local[*]") \ .config('spark.jar.packages', 'file:///home/nalla/apps/spark-3.2.4-bin-hadoop2.7/jars/spark-cassandra-connector_2.12-3.2.0.jar, file:///home/nalla/apps/spark-3.2.4-bin-hadoop2.7/jars/mysql-connector-java-8.0.33.jar') \ .config('spark.cassandra.connection.host', cassandra_host_name) \ .config('spark.cassandra.connection.port', cassandra_port) \ .config('spark.cassandra.auth.username', cassandra_username) \ .config('spark.cassandra.auth.password', cassandra_password) \ .getOrCreate() def main(): print("Starting Spark Session") spark = create_spark_session() print("Reading data from Kafka topic") # Construct a streaming dataframe that reads from Kafka topic order_df = spark.readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", kafka_bootstrap_server) \ .option("subscribe", input_topic_name) \ .option("startingOffsets", "latest") \ .load() order_df1 = order_df.selectExpr("CAST(value as STRING)", "timestamp") # Define schema for orders order_schema = StructType() \ .add("order_id", StringType()) \ .add("created_at", StringType()) \ .add("discount", StringType()) \ .add("product_id", StringType()) \ .add("quantity", StringType()) \ .add("subtotal", StringType()) \ .add("tax", StringType()) \ .add("total", StringType()) \ .add("customer_id", StringType()) order_df2 = order_df1.select(from_json(col("value"), order_schema).alias("orders"), "timestamp") order_df3 = order_df2.select("orders.*", "timestamp") print("Writing data to Cassandra") order_df3.writeStream \ .trigger(processingTime='15 seconds') \ .outputMode("update") \ .foreachBatch(save_to_cassandra) \ .start() print("Loading customer data") # Load customer data customer_df = spark.read.format('csv') \ .option('header', True) \ .option('inferSchema', True) \ .load(config.get("paths", "customer_data_file_path")) # Join with order data order_df4 = order_df3.join(customer_df, order_df3.customer_id == customer_df.customer_id, 'inner') # Aggregate data order_df5 = order_df4.groupBy("source", "state") \ .agg({'total': 'sum'}).select("source", "state", col("sum(total)").alias("total_sum_amount")) print("Writing aggregate data to console for debugging") # Write to console for debugging order_df5.writeStream \ .trigger(processingTime='15 seconds') \ .outputMode('update') \ .option("truncate", "false") \ .format("console") \ .start() print("Saving aggregate data to MySQL") # Save final results to MySQL order_df5.writeStream \ .trigger(processingTime='15 seconds') \ .outputMode('update') \ .foreachBatch(save_to_mysql) \ .start() spark.streams.awaitAnyTermination() if __name__ == "__main__": main()

Hi, Thanks for your wonderful content !, somehow actual video code is not in the downloaded resource. and save_to_mysql is not cleared could you please upload the actual content what you taught on this video, In the latest script of realtime_data_processing.py "cassandra part" is totally missing. please consider @DataMaking

thank you, bro. what version you have setup. Do you have the superset4 version.

super bro

DO you have another video, which explains how to use superset?

finally after two days of searching found best video about spark streaming of my interest! Thnak You !❤

where have you installed all the necessary software, if you have infra setup video for the same

how to setup the connection if the spark cluster is not local, there is no option to enter username and password except host and port in the UI

I am a new bee in the Data Engineering domain. I have idea on the most of the tech stacks of big data application but I was looking to connect all the dots. Your explanation was really very nice and smooth and the example also very relatable with current scenario world. Awesome content. Please keep uploading such good stuffs 🙏🙏

Sir how to save this current work are we able to continue from this step afterwards ?

i have a huge data after doing this it was 20 millions rows , which is taking lot of time

my ubuntu don´t recognize the bash comand

I'm getting the following error can somebody help hadoop@rahulparihar-Inspiron-N5010:~$ hive Hive Session ID = 661e5791-5e5d-497e-83cc-41167373ad0a Exception in thread "main" java.lang.ClassCastException: class jdk.internal.loader.ClassLoaders$AppClassLoader cannot be cast to class java.net.URLClassLoader (jdk.internal.loader.ClassLoaders$AppClassLoader and java.net.URLClassLoader are in module java.base of loader 'bootstrap') at org.apache.hadoop.hive.ql.session.SessionState.<init>(SessionState.java:413) at org.apache.hadoop.hive.ql.session.SessionState.<init>(SessionState.java:389) at org.apache.hadoop.hive.cli.CliSessionState.<init>(CliSessionState.java:60) at org.apache.hadoop.hive.cli.CliDriver.run(CliDriver.java:705) at org.apache.hadoop.hive.cli.CliDriver.main(CliDriver.java:683) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:566) at org.apache.hadoop.util.RunJar.run(RunJar.java:328) at org.apache.hadoop.util.RunJar.main(RunJar.java:241)

Majority of user uses windows so why not also do for windows installation

error despues de --verbose nano hive-site.xml <property> <name>javax.jdo.option.ConnectionURL</name> <value>jdbc:mysql://localhost/metastore?createDatabaseIfNotExist=true&amp;autoReconnect=true&amp;useSSL=false&amp;allowPublicKeyRetrieval=true</value> </property>

How to fix the error externally-managed-environment

what ip addr was it sir ?

Sir, can you kindly post the next video by showing demo for consumer and producer

Thanks for vid

Now, it's still working. Thank you for this video I can see many insteresting about set up big data tool, I will see it in the future

tks 👌

It worked for me!!!!! Thank you!

Sir, please provide the commands text file. It will be very helpfull for us.

It is saying unsupported file type when i'm trying to add cassandra rep to system packages

Hi. Your video is very good. You clearly provided the steps to be easy understand. I need your text editor which give every step. Can you upload to the subscribe link.

can you give the instruction file ?

Pin this For those of you who got an error after the --verbose part. here's the fix. First off one of the xml file he provided has 'Hive' as user and 'Datamaking' as password. Go back and change that to your username and password (your mysql user and pass) After that its likely still failing because of the permissions to retrieve key are denied. To fix that, 1. Connect to mysql cli in ur terminal: mysql -u your_username -p #enter that password 2. Once you're logged into the MySQL command-line client, execute the ALTER USER command to change the authentication plugin for the desired user. Here's the command: ALTER USER 'your_user'@'localhost' IDENTIFIED WITH mysql_native_password BY 'your_password'; 3. Thats it. Run the command again and after like 2 minutes of scrolling and generating shit, it will complete the beeline thing.

Hello, Thanks you for the series, but I cant get the source code by following the link. Can you give me anothor link such as github?

Thanks Brother, worked fine :)

Thanks for the video, and for being human! Keep making data!

Hi, I am not able to sign up into your website ,no otp is coming in mail.

I signed up to your website to get the dataaset, but i never received the confirmation email...

please share that xml code

Is there any problem if streaming data is continuously pushed to the database?

Short and very helpfull. Thank u dude!

Can you provide the names of all jar files with their versions which you used in config of saprksession

I clicked on the ressources link and it gives page not found. Did you stop your project?

Finally able to install it, thank you🎉👏