Hi All I am receiving below error while running my spark job which does insert overwrite of table. Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.protocol.DSQuotaExceededException): The DiskSpace quota of /user/anj is exceeded: quota = 3221225472 B = 3 GB but diskspace consumed = 4638218516 B = 4.32 GB at org.apache.hadoop.hdfs.server.namenode.DirectoryWithQuotaFeature.verifyStoragespaceQuota(DirectoryWithQuotaFeature.java:214) at org.apache.hadoop.hdfs.server.namenode.DirectoryWithQuotaFeature.verifyQuota(DirectoryWithQuotaFeature.java:241) at org.apache.hadoop.hdfs.server.namenode.FSDirectory.verifyQuota(FSDirectory.java:1074) at org.apache.hadoop.hdfs.server.namenode.FSDirRenameOp.verifyQuotaForRename(FSDirRenameOp.java:102) at org.apache.hadoop.hdfs.server.namenode.FSDirRenameOp.unprotectedRenameTo(FSDirRenameOp.java:402) I have configured below python code in oozie to perform insert overwrite for multiple types of files. import argparse from pyspark.sql import SparkSession import re def read_file(src_path, file_typ): df = None if file_typ.lower() == "parquet": df = spark.read.parquet(src_path) elif file_typ.lower() == "avro": df = spark.read.format("com.databricks.spark.avro").load(src_path) return df def create_query(db_nm,tab_nm,db_dir,comp_typ,file_typ): file_typ = file_typ.lower() # get the colums in string with its datatype cols_df = spark.sql("describe local_tab_" + tab_nm ) col_lst = cols_df.rdd.map(lambda x:str(x.col_name + ' ' + x.data_type)).collect() cols = ",".join(col_lst) cols = cols.lower() # query to create table query = "create table if not exists " + db_nm + "." + tab_nm + "(" + cols + ")" + " stored as " + file_typ + " location '" + db_dir + tab_nm + "' tblproperties('" + file_typ + ".compress'='" + comp_typ + "')" out = spark.sql(query) return query def insert_cols_query(db_nm,tab_nm): # get the colums in string cols_df = spark.sql("show columns in local_tab_" + tab_nm ) col_lst = cols_df.rdd.map(lambda x:str(x.col_name)).collect() cols = ",".join(col_lst) cols = cols.lower() spark.sql("set hive.exec.dynamic.partition.mode=nonstrict") # get the columns in string without the p_col column query = "insert overwrite table " + db_nm + "." + tab_nm + " select " + cols + " from local_tab_" + tab_nm out = spark.sql(query) return query def get_args(): parser = argparse.ArgumentParser() parser.add_argument("--tabname",help="table name") parser.add_argument("--fileformat",help="fileformat") args = parser.parse_args() if args.tabname: tabname = args.tabname if args.fileformat: fileformat = args.fileformat return (tabname,fileformat) if __name__ == "__main__": arg = get_args() tab_nm = arg[0] file_typ = arg[1] #### code ######## src_path="/appldigi/data/raw/landing/" + tab_nm db_nm="db_appldigi" db_dir="/appldigi/hive/db_appldigi/" comp_typ="snappy" spark = SparkSession.builder.enableHiveSupport().getOrCreate() df = read_file(src_path, file_typ) df.createOrReplaceTempView("local_tab_" + tab_nm ) chk_tab = spark.sql("show tables in " + db_nm) flag_tab_exists = chk_tab.filter(chk_tab.isTemporary=='false').filter(chk_tab.tableName==tab_nm.lower()).count() out="" if flag_tab_exists==0: # should not happened out = create_query(db_nm,tab_nm,db_dir,comp_typ,file_typ) out = insert_cols_query(db_nm,tab_nm) print out
... View more
