Handling corrupted records in spark | PySpark | Databricks 

Directly connect with me on:- topmate.io/manish_kumar25

your simply super, I am a Azure Solution Architect, but now I would like to start my journey with Data Engineering. I am very lucky that there is such very valuable and appreciable learning opportunity from your channel. Your are really good my dear, explaining concepts in good understandable way with execution. Even I am recommending your videos to my colleagues and friends

Hello Manish, I have been following your course from last couple of days and so far I have covered 17 session of Spark Theory and 7 sessions of Spark Practical. Thank You for all your efforts. Before this I have purchased multiple course on python and pyspark but I lost interest in each of the courses as they were monotonous. I'm actively looking for a job change and interviews are on pipeline, and I got the confidence on PySpark after watching your videos. Thank You ❤.

I am preparing data engineer interview from your videos 😊 Thank you You are doing great job

Very well explained Sir, Thank you.Keep educating n sharing your knowledge n experience..❤❤❤

very well explained . The way you are first ask the questions and then explain simply. Thank you so much.

Best thing about this series is potential interview questions. I can challenge you will not find this none of channels other than this in the entire RU-vid. Also we can make a separate document consisting of these questions only which will be greatly beneficial during interview preparation.

Manish bhai thank you so much for this. Mujhe jo doubts the woh apne aap clear ho rahe, really appreciate this videos bhai.

"khud Jake type Karo Ctrl + Enter marneke liar nahi phada raha huun" Thank you sir !!! ❤

Manish Bhai plz continue the same way now ur following to explain the concept. It will be Crystal clear for every one also helpful to interviews. Over all excellent THX for ur work and efforts in making content.

Jo bhai ko dataset mai issue aa raha hai bo sunhe ki manish bhai kay dataset ko ak csv mai save kare ushe csv ko notepad mai open kare waha par , aap dekhoge ki "India,Nomine" aishe likh hai ishe change kar de " " hatha de aur save karke upload kar de aapka code chalne lagega wahi results return karega manish bhai bala .

thanks Manish ....very well explained ....This spark series is very top notch

bhaia ye jo approch hai na padhane ka interview questions ke through wo bhot mst hai!🙏

Manish bhai, You are the best teacher.

Thanks Manish! I am studying late night after finishing office. Will make transition to DE soon. Thanks for the video!

Nice explanation ....All spark session are helpful for me Thanku u manish sir.

Aap data engineer ke betaj badshah ho ... please ese hi video banate rho

Thank you bro. Big Hug🙂So many things to learn

Bahut ache Manish Bhai, I am eagerly waiting for upcoming video ❤

Great interview questions ❤

Consistently following your videos, these videos are helping me a ton.

Thanks much Manish for doing this amazing job.

Manish bad record handling doc share khariya..😊

if anyone facing issue with the data just copy the data and paste the data in notepad and name the file and add .csv on it

Hi sir, Your videos are really interesting and the way you are teaching is too easy. One doubt i couldn't find the file link in the description. Could you pls provide the link. May god fulfill all your dreams. Thank you☺

When I tried the same on databricks, I'm getting the other 3 records instead of bad records.

How will spark know that it's bad records. Based on what conditions ,it's deciding it's bad ?

Nice Tutorial!!, I was trying to implement the same but found that "badRecordsPath" option in only a databricks specific feature and I was executing locally in my machine.

Hi Manish, When I run only this df = spark.read.format("csv") \ .option("inferschema", "true") \ .option("header", "true") \ .option("mode", "FAILFAST") \ .load("/FileStore/tables/data.csv") and then run df.count(), it is showing 5 records in all 3 modes. But when I am running df.show(), it is giving output as per your explanation. What can be the possible reasons for the behavior of the count function?

addicted to this channel

Manish Ji ...apka concept delivery bahaut acha hai, lekin yad karane ke liye koe document dijiye, kyo ki bar bar video dekhane me time lag raha hai....

12:13 after printing corrupt records... in my case only nomine2,nomine 3 came under the column ..while in videos whole details of id 3 and 4 came. is there any catch here..as i followed same approach

Thanks Manish please upload upcoming video ASAP

I am getting the reverted outputs..... 2 right records and 3 corrupted records

at 12:00 instead of creating whole schema can we create new column using withColumn function? or do we need to create eplicit schema to hangle bad records? could you ans?

I had a question, let's say if we have a CSV file which has some data in comma lets say address itself has commas so can we pass some text wrap in pyspark? Data can be like "Six Street,Ontario", so how can we pass this because this is not a corrupted record.

Mja aagya bhaiya, thanks

bhai....can you please provide the English subtitles too.. just to understand in better way

Hi Manish bhai, one question, When we create schema for corrupt record @11:00, I type "corrupt_record" instead of "_corrupt_record". Due to this missing prefix "_" , the corrupt records displayed to me was in different format. Can you please explain why an underscore mattered here, though it was passed in quotes as a column name? Anyone else who knows the reason can pitch in here. Schema created by you: emp_schema = StructType( [ StructField("id", IntegerType(), True), StructField("name", StringType(), True), StructField("age", IntegerType(), True), StructField("salary", IntegerType(), True), StructField("address", StringType(), True), StructField("nominee", StringType(), True), StructField("_corrupt_record", StringType(), True) ] ) Output: +---+--------+---+------+------------+--------+-------------------------------------------+ |id |name |age|salary|address |nominee |_corrupt_record | +---+--------+---+------+------------+--------+-------------------------------------------+ |1 |Manish |26 |75000 |bihar |nominee1|null | |2 |Nikita |23 |100000|uttarpradesh|nominee2|null | |3 |Pritam |22 |150000|Bangalore |India |3,Pritam,22,150000,Bangalore,India,nominee3| |4 |Prantosh|17 |200000|Kolkata |India |4,Prantosh,17,200000,Kolkata,India,nominee4| |5 |Vikash |31 |300000|null |nominee5|null | +---+--------+---+------+------------+--------+-------------------------------------------+ ========================================================================================================================================================= Schema created by me: emp_schema = StructType( [ StructField("id", IntegerType(), True), StructField("name", StringType(), True), StructField("age", IntegerType(), True), StructField("salary", IntegerType(), True), StructField("address", StringType(), True), StructField("nominee", StringType(), True), StructField("corrupt_record", StringType(), True) ] ) Output +---+--------+---+------+------------+--------+--------------+ |id |name |age|salary|address |nominee |corrupt_record| +---+--------+---+------+------------+--------+--------------+ |1 |Manish |26 |75000 |bihar |nominee1|null | |2 |Nikita |23 |100000|uttarpradesh|nominee2|null | |3 |Pritam |22 |150000|Bangalore |India |nominee3 | |4 |Prantosh|17 |200000|Kolkata |India |nominee4 | |5 |Vikash |31 |300000|null |nominee5|null | +---+--------+---+------+------------+--------+--------------+ Enjoying this playlist❤ Thanks, Sajid

At 15:32 why only 3 records are shown as the mode is permissive, shouldn't the query fetch all the records ?

Hi Manish, Nice so;ution but what if I have 500+ columns in my table, how can I do that then?

dataset bhi de dijiye Sir sath me so good hands on ho jayega

Hello Manish Hum ab tak flight_csv mein kaam kr rhe the, fir ye smployess_csv kab daalna hai pls guide.

Can we do this when reading XML and JSON files

Awesome buai

you are best

Hi @manish, After created extra column to store corrupt data, instead of getting whole row i'm getting extra values present in other column like 'nominee3'. while i was watching video i have confusion how whole row of corrupt data stored automatically.

Hi Manish, I am getting same records in all three modes. can you please help here

Where did we get this CSV file --employee data

Thank you bhai...

in permissive mode why it didn't created a new column?

corrupted data and complex data both are same or different in spark?

How to delete created table in SPARK we created?

Hi manish, Where we can check employee file

Hi Manish, If we have input csv file and we have not defined any manual schema, then in that case to show corrupted records- do we have to manually define schema for corrupted record column or how to handle that?

Manish ji, mera ek doubt hai. How corrupt_record column is taking data from the beginning(i.e. ID column)of the row?

Hi Manish, I have tried saving the corrupted records to a file, but i am unable to use %fs ls. It shows error - UsageError: Line magic function `%fs` not found. Can you help here?

Hello Guys, I was using DROPMALFORMED mode of spark but in my case corrupted data is not being removed after applying it. Is it because there is update in functionality of this option or I am doing anything wrong?

where can I get the csv file used

how do you trace why a record is corrupt , or capture error while parsing

Where can i find CSV file?

I converted the text file to CSV, but when printed that same CSV in data bricks a new column i.e. 6th column is generated with null values, so basically I am not getting 3 different table for 3 different modes, what could be possible error ?

hello sir kya as a freshers bhi this all questions are asked ?

how can we reset our databrick password. i am unable to reset .please suggest

badRecordsPath not getting created in local mode

IllegalArgumentException: If 'badRecordsPath' is specified, 'mode' is not allowed to set. mode: PermissiveMode This is the error while storing the bad record . Please advice

@Manish Kumar, in option are you using columnNameOfCorruptRecord, because otherwise it's not displaying corrupt records.. I don't know how my previous comment got deleted I searched for this option in the internet and it worked for me.

How to handle corrupted data in parquet file ?

Hello Manish Bhai.. CSV file aapne mention nhi kiya description me...please provide krenge kya

i dont see my file in fileStore

there is no csv file in description.

sir i created schema after that in my case it shows in correpted column as nominee why it is shows like this

Plz upload your next videos bhaiya 😊

Hi where is csv file link ? not able to see in description

why there is 1 job in Permissive mode 3 jobs in DROPMALFORMED and FAILFAST

Where is the spark doccumentation source?

What if we have 100 columns nd we want to print bad records...so in that case it is not possible to create schema manually...any other option to print bad records in that case????

Hi Manish. From where are you learning Scala ?

where is the CSV file ?

Sir what happens when our DF is emp_details, in this DF we have total 9 rows, 6 columns, by adding _currpet_Record column in my_schema. How it works/not?, plz xplain. Id,name,age,sal,add,nomine, 1,raju,17,15000,india,nom1 2,mani,19,21000,usa,nom2 3,Mona,21,31000,usa,nom3 4,rani,32,4100,ind,nom4 5,Mira,25,5000,mum,ind,nom5 6,yoo,21,510,mum,mh,IND,nom6 7,mahi,27,611,hyd,TS,Ind,nom7 8,nani,31,711,hyd,TS,ind,nom1,nom2 9,om,21,911,Pune,mh,ind,nom1,nom2,nom3

i m doing in jupiter noteboo i m i have run the code but data is not getting stored in new column , n=in new column i m only getting null and nomminee 1 and 2 what error or gap can be the reason of this ???

Guys, where can I find the employee file? I have been following the series from first.. not sure where is it. I am not finding data in description.

hi manish, facing problem reading the csv file when creating the csv by myself, after running the code, only showing the output, but mode fuction is not working, headers is also not reflecting, probabaly due to incorrect csv file format, can you help me in sharing your csv file so that i can download,that file ,this will help a lot.thanks

Not found csv data in description

can you please tell me how to add comments

I tried but its not shows the seventh column

Can you release your videos in english ??

where is the Data , csv file

Bhaiya next video kb ayegi

I am not able to print corrupted records, when I am printing records it forms a new column with nominee value.I am not able to understand what am I doing wrong. from pyspark.sql.types import StructType,StructField,IntegerType,StringType emp_schema=StructType([ StructField("id",IntegerType(),True), StructField("name",StringType(),True), StructField("age",IntegerType(),True), StructField("salary",IntegerType(),True), StructField("address",StringType(),True), StructField("nominee",StringType(),True), StructField("corrrecod",StringType(),True) ]) employee_df=spark.read.format("csv")\ .option("header","true")\ .option("inferschema","true")\ .option("mode","PERMISSIVE")\ .schema(emp_schema)\ .load("/FileStore/tables/EmployeeDetails.csv") employee_df.show(truncate=False)

CSV kha hai

HI @ manish_kumar_1 I tried with the data you provided but I am not able to see the corrupted records based on on different modes and when I created the employee schema and tried to see the corrupted records it is showing all the records under corrupted records.

why I am getting corrupt record here? Also corrupt record value is only nominee emp_df = spark.read.format("csv")\ .option("header", "true")\ .option("inferschema","true")\ .schema(emp_schema)\ .option("badRecordsPath","/FileStore/tables/bad_records")\ .load("/FileStore/tables/employee_file.csv") emp_df.show(truncate = False) (1) Spark Jobs emp_df:pyspark.sql.dataframe.DataFrame = [id: integer, name: string ... 5 more fields] +---+--------+---+------+---------+-------+--------------+ |id |name |age|salary|address |nominee|corrupt_record| +---+--------+---+------+---------+-------+--------------+ |3 |Pritam |22 |150000|Bangalore|India |nominee3 | |4 |Prantosh|17 |200000|Kolkata |India |nominee4 | +---+--------+---+------+---------+-------+--------------+

Unable to find spark link @manish kumar

id,name,age,salary,address,nominee 1,Manish,26,75000,bihar,nominee1 2,Nikita,23,100000,uttarpradesh,nominee2 3,Pritam,22,150000,Bangalore,India,nominee3 4,Prantosh,17,200000,Kolkata,India,nominee4 5,Vikash,31,300000,,nominee5 Employee.csv

Instead of bad records it shows me the correct one... bad_records_df=spark.read.format("json").load("/FileStore/tables/bad_records/20230610T072018/bad_records/") bad_records_df.show(truncate = False) |dbfs:/FileStore/tables/employee_df.csv|org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: 1,Manish,26,75000,bihar,nominee1 |1,Manish,26,75000,bihar,nominee1 | |dbfs:/FileStore/tables/employee_df.csv|org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: 2,Nikita,23,100000,uttarpradesh,nominee2|2,Nikita,23,100000,uttarpradesh,nominee2| |dbfs:/FileStore/tables/employee_df.csv|org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: 5,Vikash,31,300000,,nominee5 |5,Vikash,31,300000,,nominee5 |

Sir what happens when our DF is emp_details, in this DF we have total 9 rows, 6 columns, by adding _currpet_Record column in my_schema. How it works/not?, plz xplain. Id,name,age,sal,add,nomine, 1,raju,17,15000,india,nom1 2,mani,19,21000,usa,nom2 3,Mona,21,31000,usa,nom3 4,rani,32,4100,ind,nom4 5,Mira,25,5000,mum,ind,nom5 6,yoo,21,510,mum,mh,IND,nom6 7,mahi,27,611,hyd,TS,Ind,nom7 8,nani,31,711,hyd,TS,ind,nom1,nom2 9,om,21,911,Pune,mh,ind,nom1,nom2,nom3