Forum Discussion

Helper I

3 years ago

Solved

Fabric Notebook - Message Maxsize

Hi I am facing the following issue when loading data to a table using a notebook (PySpark): Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Serialized task 1218...

Need Help

AndrewWestran
3 years ago
Problem solved.
Assuming the issue was related to the large file size, I decided to 'batch' the data load for files that failed to load whole.
Updated Code:

import os
import glob
import pandas as pds
import warnings

warnings.filterwarnings("ignore", message="iteritems is deprecated")

#spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")

directory = r"/lakehouse/default/Files/C4SQLStage/DBTZBA/*.csv"

for filename in glob.glob(directory):
f = os.path.join(directory, filename)

# Load each csv file into a new Pandas DataFrame
# We use Pandas here as it supports multi-character delimiters
df = pds.read_csv(f, delimiter='', engine='python')

# Get column details for checking the number of columns
col = df.columns

# For 54 column table versions, set the column names
if len(col) != 55:
#print(f"{filename}, Columns: {len(col)}, DBName: {df.iat[0, 1]}, Date Last Updated: {df.iat[0, 53]}")
if df.iat[0, 1][0:2].upper() == 'DW':
df.columns = ['ClientID','C4SqlClientDatabaseName','cMarktNr','dtDatum','siKassNr','lBonNr','lPosNr','cStornoZle','cStorZleManag','cRueckZahlArt','cGewArt','cPfandArt','cNegArt','cStornoArt','cPreisInfo','cPreisUeb','cRabatt','cWaagArt','cDatenEing','cArtInTrAkt','cZwSumme','cPresseCode','cArtSatz','cManEing','cManBedRab','cMxNAutoRab','cPersRabatt','cAutoRabatt','cBonbezug','cKombiRab','dPluNr','dMenge','dVKPreis','dUmsatz','dSTPreis','siAbteilung','siRabSatz','dRabSumme','dArtNr','sArtBez','dEKPreis','cArtBezLang','siHauptabt','lRegalnr','siFachfolnr','lAktionsnr','lWgr','siPrzMwSt','dRabWertBon','dPreisUebWert','dVirtRabWert','lRabAktNr','lBAAutorNr','siBACodeNr']
else:
df.columns = ['cMarktNr','dtDatum','siKassNr','lBonNr','lPosNr','cStornoZle','cStorZleManag','cRueckZahlArt','cGewArt','cPfandArt','cNegArt','cStornoArt','cPreisInfo','cPreisUeb','cRabatt','cWaagArt','cDatenEing','cArtInTrAkt','cZwSumme','cPresseCode','cArtSatz','cManEing','cManBedRab','cMxNAutoRab','cPersRabatt','cAutoRabatt','cBonbezug','cKombiRab','dPluNr','dMenge','dVKPreis','dUmsatz','dSTPreis','siAbteilung','siRabSatz','dRabSumme','dArtNr','sArtBez','dEKPreis','cArtBezLang','siHauptabt','lRegalnr','siFachfolnr','lAktionsnr','lWgr','siPrzMwSt','dRabWertBon','dPreisUebWert','dVirtRabWert','lRabAktNr','lBAAutorNr','siBACodeNr','ClientID','C4SqlClientDatabaseName']
# For 55 column table versions, set the column names
else:
#print(f"{filename}, Columns: {len(col)}, DBName: {df.iat[0, 1]}, Date Last Updated: {df.iat[0, 54]}")
if df.iat[0, 1][0:2].upper() == 'DW':
df.columns = ['ClientID','C4SqlClientDatabaseName','cMarktNr','dtDatum','siKassNr','lBonNr','lPosNr','cStornoZle','cStorZleManag','cRueckZahlArt','cGewArt','cPfandArt','cNegArt','cStornoArt','cPreisInfo','cPreisUeb','cRabatt','cWaagArt','cDatenEing','cArtInTrAkt','cZwSumme','cPresseCode','cArtSatz','cManEing','cManBedRab','cMxNAutoRab','cPersRabatt','cAutoRabatt','cBonbezug','cKombiRab','dPluNr','dMenge','dVKPreis','dUmsatz','dSTPreis','siAbteilung','siRabSatz','dRabSumme','dArtNr','sArtBez','dEKPreis','cArtBezLang','siHauptabt','lRegalnr','siFachfolnr','lAktionsnr','lWgr','siPrzMwSt','dRabWertBon','dPreisUebWert','dVirtRabWert','lRabAktNr','lBAAutorNr','siBACodeNr','C4SQL_LastUpdated']
else:
df.columns = ['cMarktNr','dtDatum','siKassNr','lBonNr','lPosNr','cStornoZle','cStorZleManag','cRueckZahlArt','cGewArt','cPfandArt','cNegArt','cStornoArt','cPreisInfo','cPreisUeb','cRabatt','cWaagArt','cDatenEing','cArtInTrAkt','cZwSumme','cPresseCode','cArtSatz','cManEing','cManBedRab','cMxNAutoRab','cPersRabatt','cAutoRabatt','cBonbezug','cKombiRab','dPluNr','dMenge','dVKPreis','dUmsatz','dSTPreis','siAbteilung','siRabSatz','dRabSumme','dArtNr','sArtBez','dEKPreis','cArtBezLang','siHauptabt','lRegalnr','siFachfolnr','lAktionsnr','lWgr','siPrzMwSt','dRabWertBon','dPreisUebWert','dVirtRabWert','lRabAktNr','lBAAutorNr','siBACodeNr','ClientID','C4SqlClientDatabaseName','C4SQL_LastUpdated']

# Create a new DataFrame using the common subset of columns from the DBTZBA table
df_common = df.loc[:, ['ClientID','C4SqlClientDatabaseName','cMarktNr','dtDatum','siKassNr','lBonNr','lPosNr','cStornoZle','cStorZleManag','cRueckZahlArt','cGewArt','cPfandArt','cNegArt','cStornoArt','cPreisInfo','cPreisUeb','cRabatt','cWaagArt','cDatenEing','cArtInTrAkt','cZwSumme','cPresseCode','cArtSatz','cManEing','cManBedRab','cMxNAutoRab','cPersRabatt','cAutoRabatt','cBonbezug','cKombiRab','dPluNr','dMenge','dVKPreis','dUmsatz','dSTPreis','siAbteilung','siRabSatz','dRabSumme','dArtNr','sArtBez','dEKPreis','cArtBezLang','siHauptabt','lRegalnr','siFachfolnr','lAktionsnr','lWgr','siPrzMwSt','dRabWertBon','dPreisUebWert','dVirtRabWert','lRabAktNr','lBAAutorNr','siBACodeNr']]
try:
# Set up a spark DataFrame here to handle the write (append) to the new table
spark_df = spark.createDataFrame(df_common)
spark_df.write.mode("append").format("delta").saveAsTable("DBTZBA")
except Exception as e:
print(f"Error processing file: {filename}, Trying Month level batching.")

# Iterate through the range of months represented in dtDatum values
for unqValue in pds.to_datetime(df_common['dtDatum']).dt.strftime('%Y-%m').unique():

# Filter the DBTZBA DataFrame for the selected month
spark_df = spark.createDataFrame(df_common[pds.to_datetime(df_common['dtDatum']).dt.strftime('%Y-%m') == unqValue])

# Append the csv data to the new table
# We use spark here as it supports writing to tables
try:
spark_df.write.mode("append").format("delta").saveAsTable("DBTZBA")
except Exception as e2:
print(f"Error processing file: {filename}, Error: {e2}")

AndrewWestran

Helper I

3 years ago

Problem solved.

Assuming the issue was related to the large file size, I decided to 'batch' the data load for files that failed to load whole.

Updated Code:

import os

import glob

import pandas as pds

import warnings

warnings.filterwarnings("ignore", message="iteritems is deprecated")

#spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")

directory = r"/lakehouse/default/Files/C4SQLStage/DBTZBA/*.csv"

for filename in glob.glob(directory):

f = os.path.join(directory, filename)

# Load each csv file into a new Pandas DataFrame

# We use Pandas here as it supports multi-character delimiters

df = pds.read_csv(f, delimiter='', engine='python')

# Get column details for checking the number of columns

col = df.columns

# For 54 column table versions, set the column names

if len(col) != 55:

#print(f"{filename}, Columns: {len(col)}, DBName: {df.iat[0, 1]}, Date Last Updated: {df.iat[0, 53]}")

if df.iat[0, 1][0:2].upper() == 'DW':

df.columns = ['ClientID','C4SqlClientDatabaseName','cMarktNr','dtDatum','siKassNr','lBonNr','lPosNr','cStornoZle','cStorZleManag','cRueckZahlArt','cGewArt','cPfandArt','cNegArt','cStornoArt','cPreisInfo','cPreisUeb','cRabatt','cWaagArt','cDatenEing','cArtInTrAkt','cZwSumme','cPresseCode','cArtSatz','cManEing','cManBedRab','cMxNAutoRab','cPersRabatt','cAutoRabatt','cBonbezug','cKombiRab','dPluNr','dMenge','dVKPreis','dUmsatz','dSTPreis','siAbteilung','siRabSatz','dRabSumme','dArtNr','sArtBez','dEKPreis','cArtBezLang','siHauptabt','lRegalnr','siFachfolnr','lAktionsnr','lWgr','siPrzMwSt','dRabWertBon','dPreisUebWert','dVirtRabWert','lRabAktNr','lBAAutorNr','siBACodeNr']

else:

df.columns = ['cMarktNr','dtDatum','siKassNr','lBonNr','lPosNr','cStornoZle','cStorZleManag','cRueckZahlArt','cGewArt','cPfandArt','cNegArt','cStornoArt','cPreisInfo','cPreisUeb','cRabatt','cWaagArt','cDatenEing','cArtInTrAkt','cZwSumme','cPresseCode','cArtSatz','cManEing','cManBedRab','cMxNAutoRab','cPersRabatt','cAutoRabatt','cBonbezug','cKombiRab','dPluNr','dMenge','dVKPreis','dUmsatz','dSTPreis','siAbteilung','siRabSatz','dRabSumme','dArtNr','sArtBez','dEKPreis','cArtBezLang','siHauptabt','lRegalnr','siFachfolnr','lAktionsnr','lWgr','siPrzMwSt','dRabWertBon','dPreisUebWert','dVirtRabWert','lRabAktNr','lBAAutorNr','siBACodeNr','ClientID','C4SqlClientDatabaseName']

# For 55 column table versions, set the column names

else:

#print(f"{filename}, Columns: {len(col)}, DBName: {df.iat[0, 1]}, Date Last Updated: {df.iat[0, 54]}")

if df.iat[0, 1][0:2].upper() == 'DW':

else:

# Create a new DataFrame using the common subset of columns from the DBTZBA table

df_common = df.loc[:, ['ClientID','C4SqlClientDatabaseName','cMarktNr','dtDatum','siKassNr','lBonNr','lPosNr','cStornoZle','cStorZleManag','cRueckZahlArt','cGewArt','cPfandArt','cNegArt','cStornoArt','cPreisInfo','cPreisUeb','cRabatt','cWaagArt','cDatenEing','cArtInTrAkt','cZwSumme','cPresseCode','cArtSatz','cManEing','cManBedRab','cMxNAutoRab','cPersRabatt','cAutoRabatt','cBonbezug','cKombiRab','dPluNr','dMenge','dVKPreis','dUmsatz','dSTPreis','siAbteilung','siRabSatz','dRabSumme','dArtNr','sArtBez','dEKPreis','cArtBezLang','siHauptabt','lRegalnr','siFachfolnr','lAktionsnr','lWgr','siPrzMwSt','dRabWertBon','dPreisUebWert','dVirtRabWert','lRabAktNr','lBAAutorNr','siBACodeNr']]

try:

# Set up a spark DataFrame here to handle the write (append) to the new table

spark_df = spark.createDataFrame(df_common)

spark_df.write.mode("append").format("delta").saveAsTable("DBTZBA")

except Exception as e:

print(f"Error processing file: {filename}, Trying Month level batching.")

# Iterate through the range of months represented in dtDatum values

for unqValue in pds.to_datetime(df_common['dtDatum']).dt.strftime('%Y-%m').unique():

# Filter the DBTZBA DataFrame for the selected month

spark_df = spark.createDataFrame(df_common[pds.to_datetime(df_common['dtDatum']).dt.strftime('%Y-%m') == unqValue])

# Append the csv data to the new table

# We use spark here as it supports writing to tables

try:

spark_df.write.mode("append").format("delta").saveAsTable("DBTZBA")

except Exception as e2:

print(f"Error processing file: {filename}, Error: {e2}")

Forum Discussion

Fabric Notebook - Message Maxsize

Related content

Error message on notebook sql

error message

notebook parameter

PYSPARK notebook issue

Small notebook activity is very slow