java.lang.ArithmeticException: long overflow Exception while writing to table | pyspark

Divyanshu — Fri, 29 Sep 2023 04:39:11 GMT

Hey ,

I am trying to fetch data from mongo and write to databricks table.

I have read data from mongo using pymongo library, then flattened nested struct objects along with renaming columns(since there were few duplicates) and then writing to databricks table. Upon writing databricks throws following error:

"org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 12.0 failed 4 times, most recent failure: Lost task 2.3 in stage 12.0 (TID 53) (10.2.64.17 executor 0): org.apache.spark.SparkRuntimeException: Error while encoding: java.lang.ArithmeticException: long overflow"

I am getting same error for 197/492 columns which includes string, double, boolean, integer.

Any leads to what should I look into? Normally all answers I could find over internet are related to storing long data in int but our case seems to be very different.

Below is code and schema sample:

#Function to flatten nested json struct def flatten_json_df(_df: DataFrame) -> DataFrame: # List to hold the dynamically generated column names flattened_col_list = [] # Inner method to iterate over Data Frame to generate the column list def get_flattened_cols(df: DataFrame, struct_col: str = None) -> None: for col in df.columns: if df.schema[col].dataType.typeName() != 'struct': if struct_col is None: flattened_col_list.append(f"{col} as {col.replace('.','_')}") else: t = struct_col + "." + col flattened_col_list.append(f"{t} as {t.replace('.','_')}") else: chained_col = struct_col +"."+ col if struct_col is not None else col get_flattened_cols(df.select(col+".*"), chained_col) get_flattened_cols(_df) i=0 for col in flattened_col_list: #Removing/quoting special charachters col=re.sub(r'([0-9]+-[0-9]+) as',r'`\1` as',col) col=re.sub(r'([0-9]+)-([0-9]+)$',r'\1_\2',col) col=re.sub(r'\.(.*[0-9]+, [0-9]+) as',r'.`\1` as',col) flattened_col_list[i]=re.sub(r' ([0-9]+), ([0-9]+)$',r'_\1__\2',col) i=i+1 return _df.selectExpr(flattened_col_list) #Function to write data def new_write_delta(type,df1,tgt_db_name,tgt_table_name,full_databricks_table_name,pk,operation=""): id_column = pk tgt_db_name = tgt_db_name targetTable = tgt_table_name delta_path = "" df1=df1.dropDuplicates() pk_arr = [pk_val for pk_val in pk.split(",") if pk_val != ""] if len(pk_arr) != 0: df1=df1.dropDuplicates(pk.split(",")) if type == 'test': path = 'abfss://ft-data-container@ftdatastorage.dfs.core.windows.net/test' delta_path = path+'/'+tgt_db_name+'/'+targetTable print(f"{full_databricks_table_name} Start writing in overwrite.....") print("Writting in the location of :") print(f"{destination_table_location} in the container") ( df1 .write .mode('overwrite') .format('delta') .option('path',destination_table_location) .saveAsTable(f"{full_databricks_table_name}") ) type="test" full_databricks_table_name = tgt_db_name + '.' + 'mongo_test_' + table query = """[{ $match: { "lastUpdated" : { $gte : {$date : "%s"} } } }]"""%(start_date) spark.conf.set('spark.sql.caseSensitive', True) df = spark.read.format("mongo")\ .option("database", database)\ .option("spark.mongodb.input.uri", connectionString)\ .option("collection",table)\ .option("pipeline", query)\ .load() dfnew1=df.withColumn("wayPoint_value",explode("wayPoint")).withColumn("origin_value",df["origin"]).withColumn("destination_value",df["destination"]) #Exploding array column and rename dfnew=dfnew1.drop("wayPoint").drop("origin").drop("destination") # Removing original columns which were duplicated in above line dfnew=flatten_json_df(dfnew) #Flattening schema dfnew=dfnew.drop("destination_value_TAT").drop("origin_value_TAT").drop("wayPoint_value_TAT") # Dropping duplicate columns pk = '_id_oid' df=dfnew tgt_db_name = tgt_db_name tgt_table_name = "mongo_test_" + table #Calling write function defined above new_write_delta(type,df,tgt_db_name,tgt_table_name,full_databricks_table_name,pk)

Schema:

topic java.lang.ArithmeticException: long overflow Exception while writing to table | pyspark in Data Engineering

java.lang.ArithmeticException: long overflow Exception while writing to table | pyspark

Re: java.lang.ArithmeticException: long overflow Exception while writing to table | pyspark